diff --git a/CHANGELOG.md b/CHANGELOG.md index db9b15515..a4fb74a87 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -31,6 +31,32 @@ Notable changes between releases. Detailed migration notes for storage transitio ### Fixed +- **On queue storage, a queue whose enqueue lane is busy no longer holds + back promotion of other queues, nor the maintenance leader's rescue, + rotation and prune.** Promotion re-enqueues due scheduled and retryable + rows into their lane, which takes the lane's reservation lock that + producers hold until they commit. A promotion batch was one transaction + across every queue it touched, and the leader runs its branches one after + another, so a lane kept busy by producers delayed due jobs of unrelated + queues in the same batch, and rescue started seconds late. Promotion now + gives each `(queue, priority)` lane its own transaction under a 50 ms + `lock_timeout`. A lane that times out keeps its rows deferred and is + handed to one dedicated task per lane, shared by scheduled and retryable + rows. The task alternates states within its 32-batch budget and waits up + to 30 s for each lock acquisition; the leader skips the lane meanwhile. + Several finite producer transactions queued ahead can exhaust that wait. + A timed-out waiter releases its queue position and retries on a later + tick, so sustained contention can still prevent promotion. Each waiter + holds a pool connection, so the leader runs one per eight pool + connections, at least one and at most four; a further contended lane is + retried on the next tick. Batch selection skips rows another transaction + has locked, so locked due rows cannot fill every batch of a tick. `awa.maintenance.promote.lane_contention` + (`awa_maintenance_promote_lane_contention_total{awa_job_state,awa_job_queue,awa_promote_outcome}`) + counts `handed_off`, `waiter_limit` and `waiter_timed_out`. Promoted rows + now take lane sequence numbers in due order within each state and lane. + `QueueStorage::promote_due` keeps its signature and promotes lane by lane + without a lock timeout; `promote_due_lanes` and `promote_due_lane` expose + the per-lane form. - **A queue lane is usable by every runtime role, whichever role creates it first (migration v053).** A lane's enqueue and claim sequences belonged to the role that first enqueued into, claimed from or promoted into it, so with diff --git a/Cargo.lock b/Cargo.lock index 2055d2754..a15535ce8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -389,6 +389,7 @@ dependencies = [ "serde_json", "sqlx", "tokio", + "tokio-util", "tonic", "tracing", "tracing-opentelemetry", diff --git a/awa-metrics/src/lib.rs b/awa-metrics/src/lib.rs index 5241d6740..9dc0cfd8e 100644 --- a/awa-metrics/src/lib.rs +++ b/awa-metrics/src/lib.rs @@ -70,6 +70,7 @@ pub mod names { 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"; + pub const MAINTENANCE_PROMOTE_LANE_CONTENTION: &str = "awa.maintenance.promote.lane_contention"; pub const MAINTENANCE_BRANCH_DURATION: &str = "awa.maintenance.branch.duration"; pub const MAINTENANCE_BRANCH_OVERRUN: &str = "awa.maintenance.branch.overrun"; pub const MAINTENANCE_ROTATE_ATTEMPTS: &str = "awa.maintenance.rotate.attempts"; @@ -192,6 +193,11 @@ pub struct AwaMetrics { pub promotion_batch_size: Histogram, /// Promotion query duration. pub promotion_duration_seconds: Histogram, + /// Promotion lanes whose locks were busy past the leader's wait budget, + /// attributed to (state, queue, outcome): `handed_off` to a dedicated + /// waiter, `waiter_limit` when every waiter was busy, `waiter_timed_out` + /// when a waiter's own budget ran out. + pub promotion_lane_contention: Counter, /// Current in-flight jobs (can go up and down). pub jobs_in_flight: UpDownCounter, /// Total heartbeat batches sent. @@ -482,6 +488,11 @@ impl AwaMetrics { .with_unit("s") .with_boundaries(LATENCY_BUCKETS_SECONDS.to_vec()) .build(), + promotion_lane_contention: meter + .u64_counter(names::MAINTENANCE_PROMOTE_LANE_CONTENTION) + .with_description("Promotion lanes whose locks were busy, by state/queue/outcome") + .with_unit("{lane}") + .build(), jobs_in_flight: meter .i64_up_down_counter(names::JOB_IN_FLIGHT) .with_description("Current number of in-flight jobs") @@ -909,6 +920,23 @@ impl AwaMetrics { .record(duration.as_secs_f64(), &attrs); } + /// Record a promotion lane whose locks were busy past the leader's wait + /// budget, with what became of it (`handed_off`, `waiter_limit`, + /// `waiter_timed_out`). + pub fn record_promotion_lane_contention( + &self, + state: &str, + queue: &str, + outcome: &'static str, + ) { + let attrs = [ + opentelemetry::KeyValue::new("awa.job.state", state.to_string()), + opentelemetry::KeyValue::new("awa.job.queue", queue.to_string()), + opentelemetry::KeyValue::new("awa.promote.outcome", outcome), + ]; + self.promotion_lane_contention.add(1, &attrs); + } + /// Record in-flight change. pub fn record_in_flight_change(&self, queue: &str, delta: i64) { let attrs = [opentelemetry::KeyValue::new( @@ -1462,6 +1490,65 @@ mod tests { assert!(saw_already_pruned); } + #[test] + fn promotion_lane_contention_counts_each_outcome_per_state_and_queue() { + let exporter = InMemoryMetricExporter::default(); + let provider = SdkMeterProvider::builder() + .with_periodic_exporter(exporter.clone()) + .build(); + let metrics = AwaMetrics::new(&provider.meter("awa-metrics-test")); + + metrics.record_promotion_lane_contention("scheduled", "emails", "handed_off"); + metrics.record_promotion_lane_contention("scheduled", "emails", "handed_off"); + metrics.record_promotion_lane_contention("retryable", "emails", "waiter_limit"); + metrics.record_promotion_lane_contention("scheduled", "billing", "waiter_timed_out"); + provider.force_flush().expect("flush metrics"); + let resource_metrics = exporter.get_finished_metrics().expect("read metrics"); + + let mut counts = BTreeMap::new(); + for resource in &resource_metrics { + for scope in resource.scope_metrics() { + for metric in scope.metrics() { + if let ( + names::MAINTENANCE_PROMOTE_LANE_CONTENTION, + AggregatedMetrics::U64(MetricData::Sum(sum)), + ) = (metric.name(), metric.data()) + { + for point in sum.data_points() { + let label = |key: &str| { + point + .attributes() + .find(|kv| kv.key.as_str() == key) + .map(|kv| kv.value.as_str().into_owned()) + .unwrap_or_default() + }; + counts.insert( + ( + label("awa.job.state"), + label("awa.job.queue"), + label("awa.promote.outcome"), + ), + point.value(), + ); + } + } + } + } + } + + let key = |state: &str, queue: &str, outcome: &str| { + (state.to_string(), queue.to_string(), outcome.to_string()) + }; + assert_eq!( + counts, + BTreeMap::from([ + (key("retryable", "emails", "waiter_limit"), 1), + (key("scheduled", "billing", "waiter_timed_out"), 1), + (key("scheduled", "emails", "handed_off"), 2), + ]) + ); + } + #[test] fn prune_attempts_count_every_outcome_per_ring() { use awa_model::{PruneDurations, PruneOutcome, SkipReason}; diff --git a/awa-model/src/lib.rs b/awa-model/src/lib.rs index 8edc4f099..1d745187d 100644 --- a/awa-model/src/lib.rs +++ b/awa-model/src/lib.rs @@ -54,9 +54,9 @@ pub use partitioned_queue::{ partition_for_ordering_key, partition_hash64, PartitionedQueue, PartitionedQueueError, }; pub use queue_storage::{ - ClaimedEntry, ClaimedRuntimeJob, PruneDurations, PruneOutcome, QueueCounts, QueueStorage, - QueueStorageConfig, RingLedgerFoldOutcome, RingRotationCadence, RotateOutcome, SkipReason, - TerminalDeltaRollupOutcome, TerminalRollupFoldOutcome, + ClaimedEntry, ClaimedRuntimeJob, PromoteOutcome, PromotionLane, PruneDurations, PruneOutcome, + QueueCounts, QueueStorage, QueueStorageConfig, RingLedgerFoldOutcome, RingRotationCadence, + RotateOutcome, SkipReason, TerminalDeltaRollupOutcome, TerminalRollupFoldOutcome, }; pub use storage::StorageStatus; diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index 034e203de..135036b03 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -641,6 +641,29 @@ pub struct RingLedgerFoldOutcome { pub skipped_mvcc_pinned: bool, } +/// A `(queue, priority)` enqueue lane as promotion addresses it. Promotion +/// picks the shard when it re-enqueues a row, so the shard is not part of it. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct PromotionLane { + pub queue: String, + pub priority: i16, +} + +/// Outcome of [`QueueStorage::promote_due_lanes`] and +/// [`QueueStorage::promote_due_lane`]. +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub struct PromoteOutcome { + /// Rows moved into the ready ring. + pub promoted: usize, + /// Due rows the batch selected, including those left deferred on a + /// contended lane. A batch that selected its full size may have left + /// further due rows behind. + pub selected: usize, + /// Lanes whose promotion transaction hit `lock_timeout` and rolled back. + /// Their rows stay deferred. + pub contended: Vec, +} + /// Discriminator for [`PruneOutcome::SkippedActive`]. /// /// Multiple gates can fire `SkippedActive` for the same ring (e.g. queue @@ -699,7 +722,7 @@ fn is_lock_contention_error(err: &sqlx::Error) -> bool { ) } -async fn set_prune_lock_timeout_tx( +async fn set_lock_timeout_tx( tx: &mut sqlx::Transaction<'_, Postgres>, timeout: Duration, ) -> Result<(), AwaError> { @@ -3257,6 +3280,15 @@ struct ExistingReadyRow { payload: serde_json::Value, } +/// Which of a lane's due rows one promotion transaction takes. +#[derive(Debug, Clone, Copy)] +enum DueLaneRows<'a> { + /// These rows, picked by a batch across lanes, where still due. + Selected(&'a [i64]), + /// The oldest due rows, up to this many. + Oldest(i64), +} + #[derive(Debug, Clone, sqlx::FromRow)] struct DeletedLeaseRow { ready_slot: i32, @@ -18510,32 +18542,212 @@ impl QueueStorage { } } + /// Promote up to `batch_size` due `state` rows into the ready ring. + /// Equivalent to [`Self::promote_due_lanes`] with no lock timeout and no + /// skipped lanes, returning the number of rows promoted. pub async fn promote_due( &self, pool: &PgPool, state: JobState, batch_size: i64, ) -> Result { + Ok(self + .promote_due_lanes(pool, state, batch_size, None, &[]) + .await? + .promoted) + } + + /// Select up to `batch_size` due `state` rows, oldest `run_at` first, + /// outside the lanes in `skip`, and promote each lane's share in its own + /// transaction. + /// + /// The selection skips rows another transaction has locked, such as a + /// lane promotion still waiting for its lane, so they cannot fill every + /// batch of a tick. It runs as its own statement, so its row locks end + /// with it; `deferred_jobs` is not partitioned, so no run-time pruning + /// meets those locks. + /// + /// Re-enqueueing a row takes its lane's enqueue reservation lock, which + /// producers hold until they commit, so a lane under heavy enqueue load + /// can keep its promotion waiting. A transaction per lane confines that + /// wait to the lane's own rows. With `lane_lock_timeout`, each lane's + /// transaction runs under that `lock_timeout`; a lane that hits it rolls + /// back, keeps its rows deferred and is reported in + /// [`PromoteOutcome::contended`], and the remaining lanes go ahead. + pub async fn promote_due_lanes( + &self, + pool: &PgPool, + state: JobState, + batch_size: i64, + lane_lock_timeout: Option, + skip: &[PromotionLane], + ) -> Result { if !matches!(state, JobState::Scheduled | JobState::Retryable) || batch_size <= 0 { - return Ok(0); + return Ok(PromoteOutcome::default()); } let schema = self.schema(); + let due: Vec<(i64, String, i16)> = sqlx::query_as(audited_sql(format!( + r#" + SELECT job_id, queue, priority + FROM {schema}.deferred_jobs AS deferred + WHERE state = $1 + AND run_at <= clock_timestamp() + AND NOT EXISTS ( + SELECT 1 FROM awa.queue_meta + WHERE queue = deferred.queue AND paused = TRUE + ) + AND NOT EXISTS ( + SELECT 1 + FROM unnest($3::text[], $4::smallint[]) AS skip (queue, priority) + WHERE skip.queue = deferred.queue AND skip.priority = deferred.priority + ) + ORDER BY run_at ASC, priority ASC, job_id ASC + LIMIT $2 + FOR UPDATE OF deferred SKIP LOCKED + "# + ))) + .bind(state) + .bind(batch_size) + .bind( + skip.iter() + .map(|lane| lane.queue.as_str()) + .collect::>(), + ) + .bind(skip.iter().map(|lane| lane.priority).collect::>()) + .fetch_all(pool) + .await + .map_err(map_sqlx_error)?; + + let mut outcome = PromoteOutcome { + selected: due.len(), + ..PromoteOutcome::default() + }; + let mut lanes: BTreeMap> = BTreeMap::new(); + for (job_id, queue, priority) in due { + lanes + .entry(PromotionLane { queue, priority }) + .or_default() + .push(job_id); + } + for (lane, job_ids) in lanes { + match self + .promote_lane( + pool, + state, + &lane, + DueLaneRows::Selected(&job_ids), + lane_lock_timeout, + ) + .await? + { + Some(promoted) => outcome.promoted += promoted, + None => outcome.contended.push(lane), + } + } + Ok(outcome) + } + + /// Promote up to `batch_size` due `state` rows of one lane, oldest + /// `run_at` first, in one transaction. With `lane_lock_timeout`, a + /// transaction that hits it rolls back and the lane is returned in + /// [`PromoteOutcome::contended`] with nothing selected. + pub async fn promote_due_lane( + &self, + pool: &PgPool, + state: JobState, + lane: &PromotionLane, + batch_size: i64, + lane_lock_timeout: Option, + ) -> Result { + if !matches!(state, JobState::Scheduled | JobState::Retryable) || batch_size <= 0 { + return Ok(PromoteOutcome::default()); + } + Ok( + match self + .promote_lane( + pool, + state, + lane, + DueLaneRows::Oldest(batch_size), + lane_lock_timeout, + ) + .await? + { + Some(promoted) => PromoteOutcome { + promoted, + selected: promoted, + contended: Vec::new(), + }, + None => PromoteOutcome { + contended: vec![lane.clone()], + ..PromoteOutcome::default() + }, + }, + ) + } + + /// One lane's promotion transaction. `None` when it hit `lock_timeout` + /// and rolled back. + async fn promote_lane( + &self, + pool: &PgPool, + state: JobState, + lane: &PromotionLane, + rows: DueLaneRows<'_>, + lock_timeout: Option, + ) -> Result, AwaError> { let mut tx = pool.begin().await.map_err(map_sqlx_error)?; - let moved: Vec = sqlx::query_as(audited_sql(format!( + match self + .promote_lane_tx(&mut tx, state, lane, rows, lock_timeout) + .await + { + Ok(promoted) => match tx.commit().await { + Ok(()) => Ok(Some(promoted)), + Err(err) if is_lock_contention_error(&err) => Ok(None), + Err(err) => Err(map_sqlx_error(err)), + }, + Err(AwaError::Database(err)) if is_lock_contention_error(&err) => { + tx.rollback().await.map_err(map_sqlx_error)?; + Ok(None) + } + Err(err) => Err(err), + } + } + + async fn promote_lane_tx( + &self, + tx: &mut sqlx::Transaction<'_, Postgres>, + state: JobState, + lane: &PromotionLane, + rows: DueLaneRows<'_>, + lock_timeout: Option, + ) -> Result { + if let Some(timeout) = lock_timeout { + set_lock_timeout_tx(tx, timeout).await?; + } + + let schema = self.schema(); + let (row_filter, limit) = match rows { + DueLaneRows::Selected(_) => ("AND job_id = ANY($4)", ""), + DueLaneRows::Oldest(_) => ("", "ORDER BY run_at ASC, job_id ASC LIMIT $4"), + }; + let query = sqlx::query_as::<_, DeferredJobRow>(audited_sql(format!( r#" DELETE FROM {schema}.deferred_jobs WHERE job_id IN ( SELECT job_id - FROM {schema}.deferred_jobs + FROM {schema}.deferred_jobs AS deferred WHERE state = $1 + AND queue = $2 + AND priority = $3 + {row_filter} AND run_at <= clock_timestamp() AND NOT EXISTS ( SELECT 1 FROM awa.queue_meta - WHERE queue = {schema}.deferred_jobs.queue AND paused = TRUE + WHERE queue = deferred.queue AND paused = TRUE ) - ORDER BY run_at ASC, priority ASC, job_id ASC - LIMIT $2 + {limit} FOR UPDATE SKIP LOCKED ) RETURNING @@ -18558,19 +18770,23 @@ impl QueueStorage { "# ))) .bind(state) - .bind(batch_size) - .fetch_all(tx.as_mut()) - .await - .map_err(map_sqlx_error)?; - + .bind(&lane.queue) + .bind(lane.priority); + let query = match rows { + DueLaneRows::Selected(job_ids) => query.bind(job_ids), + DueLaneRows::Oldest(batch_size) => query.bind(batch_size), + }; + let mut moved: Vec = + query.fetch_all(tx.as_mut()).await.map_err(map_sqlx_error)?; if moved.is_empty() { - tx.commit().await.map_err(map_sqlx_error)?; return Ok(0); } + // DELETE ... RETURNING has no order; lane sequence follows due order. + moved.sort_unstable_by_key(|row| (row.run_at, row.job_id)); + let promoted = moved.len(); let ready_rows: Vec = moved - .iter() - .cloned() + .into_iter() .map(|row| ExistingReadyRow { job_id: row.job_id, kind: row.kind, @@ -18588,15 +18804,10 @@ impl QueueStorage { payload: row.payload, }) .collect(); - let queues = ready_rows - .iter() - .map(|row| row.queue.clone()) - .collect::>(); - self.insert_existing_ready_rows_tx(&mut tx, ready_rows, Some(state)) + self.insert_existing_ready_rows_tx(tx, ready_rows, Some(state)) .await?; - self.notify_queues_tx(&mut tx, queues).await?; - tx.commit().await.map_err(map_sqlx_error)?; - Ok(moved.len()) + self.notify_queues_tx(tx, vec![lane.queue.clone()]).await?; + Ok(promoted) } async fn relation_has_rows_tx( @@ -19458,7 +19669,7 @@ impl QueueStorage { } let locked: Result = async { - set_prune_lock_timeout_tx(&mut tx, self.prune_lock_timeout).await?; + set_lock_timeout_tx(&mut tx, self.prune_lock_timeout).await?; sqlx::query(audited_sql(format!( "LOCK TABLE {delta_child} IN ACCESS EXCLUSIVE MODE" @@ -19916,7 +20127,7 @@ impl QueueStorage { } let locked: Result = async { - set_prune_lock_timeout_tx(&mut tx, self.prune_lock_timeout).await?; + set_lock_timeout_tx(&mut tx, self.prune_lock_timeout).await?; let lock_started = Instant::now(); // Claims read segments before ready rows, while producers write in @@ -20316,7 +20527,7 @@ impl QueueStorage { let lease_child = lease_child_name(schema, slot as usize); let locked: Result = async { - set_prune_lock_timeout_tx(&mut tx, self.prune_lock_timeout).await?; + set_lock_timeout_tx(&mut tx, self.prune_lock_timeout).await?; let lock_started = Instant::now(); sqlx::query(audited_sql(format!( @@ -20681,7 +20892,7 @@ impl QueueStorage { } let locked: Result = async { - set_prune_lock_timeout_tx(&mut tx, self.prune_lock_timeout).await?; + set_lock_timeout_tx(&mut tx, self.prune_lock_timeout).await?; let lock_started = Instant::now(); sqlx::query(audited_sql(format!( diff --git a/awa-worker/src/maintenance.rs b/awa-worker/src/maintenance.rs index f1dfe9f79..a7b74dcfb 100644 --- a/awa-worker/src/maintenance.rs +++ b/awa-worker/src/maintenance.rs @@ -1,4 +1,5 @@ use crate::executor::DlqPolicy; +use crate::metrics::AwaMetrics; use crate::runtime::InFlightMap; use crate::storage::{QueueStorageRuntime, RuntimeStorage}; use awa_model::audited_sql; @@ -8,7 +9,8 @@ use awa_model::cron::{ #[cfg(test)] use awa_model::SkipReason; use awa_model::{ - JobRow, JobState, PeriodicJob, PruneOutcome, RotateOutcome, TerminalDeltaRollupOutcome, + AwaError, JobRow, JobState, PeriodicJob, PromotionLane, PruneOutcome, QueueStorage, + RotateOutcome, TerminalDeltaRollupOutcome, }; use chrono::Utc; use croner::Cron; @@ -18,7 +20,7 @@ use std::collections::{HashMap, HashSet}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; -use tokio::task::JoinHandle; +use tokio::task::{JoinHandle, JoinSet}; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, warn}; use uuid::Uuid; @@ -763,6 +765,246 @@ const CRON_CATCH_UP_LIMIT: usize = 1_000; const TERMINAL_COUNT_ROLLUP_MAX_SLOTS_PER_TICK: usize = 4; type QueueStorageMetricRow = (String, i64, i64, i64, i64, i64, i64, i64, Option); +/// How long a promotion transaction waits on one lane's locks before the +/// leader hands the lane to a dedicated waiter and goes on without it. +const PROMOTE_LANE_LOCK_TIMEOUT: Duration = Duration::from_millis(50); +/// Lock wait of a dedicated lane waiter. The combined wait behind several +/// transactions can exhaust this budget; a later retry gets a new queue position. +const PROMOTE_LANE_WAITER_LOCK_TIMEOUT: Duration = Duration::from_secs(30); +/// Most lanes with a dedicated waiter at once. A waiter holds a pool +/// connection while it waits, so smaller pools get fewer: one per eight +/// connections, with at least one available for contended lanes. +const PROMOTE_MAX_LANE_WAITERS: usize = 4; + +fn lane_waiter_limit(pool_max_connections: u32) -> usize { + (pool_max_connections as usize / 8).clamp(1, PROMOTE_MAX_LANE_WAITERS) +} + +/// Queue-storage promotion of due scheduled and retryable rows, as the +/// maintenance leader runs it for one store. +/// +/// Each lane's share of a batch is promoted in its own transaction under a +/// short `lock_timeout`, so a lane whose enqueue lock is busy delays only its +/// own rows. A lane that times out keeps its rows deferred and is handed to a +/// dedicated waiter task, which alternates scheduled and retryable batches +/// under a bounded lock timeout while the leader carries on with other lanes +/// and other maintenance branches. Exhausting that timeout leaves the lane +/// deferred for a later retry; sustained contention can prevent progress. +/// Dropping the promoter aborts its waiters. +#[doc(hidden)] +pub struct QueueStoragePromoter { + pool: PgPool, + store: Arc, + metrics: AwaMetrics, + waiting: Arc>>, + waiters: std::sync::Mutex>, + max_waiters: usize, +} + +impl QueueStoragePromoter { + pub fn new(pool: PgPool, store: Arc, metrics: AwaMetrics) -> Self { + let max_waiters = lane_waiter_limit(pool.options().get_max_connections()); + Self { + max_waiters, + pool, + store, + metrics, + waiting: Arc::default(), + waiters: std::sync::Mutex::new(JoinSet::new()), + } + } + + /// One promotion tick for `state`: batches until one selects fewer rows + /// than the batch size, each skipping lanes that have a waiter or were + /// contended earlier in the tick. Returns the rows promoted by the tick + /// itself, not by waiters. + pub async fn promote_due( + &self, + state: JobState, + cancel: &CancellationToken, + ) -> Result { + { + let mut waiters = self.waiters.lock().expect("lane waiters poisoned"); + while let Some(result) = waiters.try_join_next() { + if let Err(error) = result { + error!(%error, "Promotion lane waiter task failed"); + } + } + } + + let mut contended: Vec = Vec::new(); + let mut promoted = 0usize; + for _ in 0..PROMOTE_MAX_BATCHES_PER_TICK { + if cancel.is_cancelled() { + break; + } + let mut skip = self.waiting_lanes(); + skip.extend(contended.iter().cloned()); + let batch_started = Instant::now(); + let outcome = self + .store + .promote_due_lanes( + &self.pool, + state, + PROMOTE_BATCH_SIZE, + Some(PROMOTE_LANE_LOCK_TIMEOUT), + &skip, + ) + .await?; + self.metrics.record_promotion_batch( + state.as_str(), + outcome.promoted as u64, + batch_started.elapsed(), + ); + promoted += outcome.promoted; + for lane in outcome.contended { + self.hand_off(state, lane.clone()); + contended.push(lane); + } + if (outcome.selected as i64) < PROMOTE_BATCH_SIZE { + break; + } + } + Ok(promoted) + } + + fn waiting_lanes(&self) -> Vec { + self.waiting + .lock() + .expect("waiting lanes poisoned") + .iter() + .cloned() + .collect() + } + + fn hand_off(&self, state: JobState, lane: PromotionLane) { + { + let mut waiting = self.waiting.lock().expect("waiting lanes poisoned"); + if waiting.contains(&lane) { + return; + } + if waiting.len() >= self.max_waiters { + self.metrics.record_promotion_lane_contention( + state.as_str(), + &lane.queue, + "waiter_limit", + ); + debug!( + queue = %lane.queue, + priority = lane.priority, + "Promotion lane contended with every lane waiter busy; retrying next tick" + ); + return; + } + waiting.insert(lane.clone()); + } + self.metrics + .record_promotion_lane_contention(state.as_str(), &lane.queue, "handed_off"); + let guard = WaitingLane { + waiting: Arc::clone(&self.waiting), + lane, + }; + let pool = self.pool.clone(); + let store = Arc::clone(&self.store); + let metrics = self.metrics.clone(); + self.waiters + .lock() + .expect("lane waiters poisoned") + .spawn(async move { + promote_waiting_lane(&pool, &store, &metrics, state, &guard.lane).await; + drop(guard); + }); + } +} + +/// Holds a lane in [`QueueStoragePromoter`]'s waiting set until its waiter +/// finishes, panics or is aborted. +struct WaitingLane { + waiting: Arc>>, + lane: PromotionLane, +} + +impl Drop for WaitingLane { + fn drop(&mut self) { + if let Ok(mut waiting) = self.waiting.lock() { + waiting.remove(&self.lane); + } + } +} + +async fn promote_waiting_lane( + pool: &PgPool, + store: &QueueStorage, + metrics: &AwaMetrics, + initial_state: JobState, + lane: &PromotionLane, +) { + let states = match initial_state { + JobState::Scheduled => [JobState::Scheduled, JobState::Retryable], + JobState::Retryable => [JobState::Retryable, JobState::Scheduled], + _ => return, + }; + let mut short_batches = [false; 2]; + for batch in 0..PROMOTE_MAX_BATCHES_PER_TICK { + let state_index = batch % states.len(); + let state = states[state_index]; + let batch_started = Instant::now(); + match store + .promote_due_lane( + pool, + state, + lane, + PROMOTE_BATCH_SIZE, + Some(PROMOTE_LANE_WAITER_LOCK_TIMEOUT), + ) + .await + { + Ok(outcome) if !outcome.contended.is_empty() => { + metrics.record_promotion_lane_contention( + state.as_str(), + &lane.queue, + "waiter_timed_out", + ); + warn!( + queue = %lane.queue, + priority = lane.priority, + lock_timeout_ms = PROMOTE_LANE_WAITER_LOCK_TIMEOUT.as_millis() as u64, + "Promotion of a contended lane timed out waiting for its locks; retrying next tick" + ); + return; + } + Ok(outcome) => { + metrics.record_promotion_batch( + state.as_str(), + outcome.promoted as u64, + batch_started.elapsed(), + ); + short_batches[state_index] = (outcome.selected as i64) < PROMOTE_BATCH_SIZE; + if short_batches.iter().all(|short| *short) { + return; + } + } + Err(err) => { + error!( + queue = %lane.queue, + priority = lane.priority, + error = %err, + "Failed to promote a contended lane" + ); + return; + } + } + } +} + +/// The leader's queue-storage promoters: its own store's, and the configured +/// store a canonical-resolved leader promotes once routing has flipped +/// (#456). Scoped to one leadership term. +struct LeaderPromoters { + primary: Option, + standby: Option, +} + impl MaintenanceService { #[allow(clippy::too_many_arguments)] pub(crate) fn new( @@ -1088,6 +1330,22 @@ impl MaintenanceService { // tick. Local to the leader loop so a re-election resets // backoff to zero. let prune_tracker = PruneBackoffTracker::new(); + let promoters = LeaderPromoters { + primary: self.storage.queue_storage().map(|runtime| { + QueueStoragePromoter::new( + self.pool.clone(), + Arc::clone(&runtime.store), + self.metrics.clone(), + ) + }), + standby: self.standby_queue_storage.as_ref().map(|runtime| { + QueueStoragePromoter::new( + self.pool.clone(), + Arc::clone(&runtime.store), + self.metrics.clone(), + ) + }), + }; // Skip the first immediate tick heartbeat_rescue_timer.tick().await; @@ -1153,7 +1411,7 @@ impl MaintenanceService { } _ = promote_timer.tick() => { if let Some(timer) = branch_tracker.try_begin_without_cooldown("promote_scheduled", self.promote_interval, &self.metrics) { - self.promote_scheduled().await; + self.promote_scheduled(&promoters).await; timer.finish(); } } @@ -2541,13 +2799,20 @@ impl MaintenanceService { } /// Promote scheduled jobs that are now due. - #[tracing::instrument(skip(self), name = "maintenance.promote")] - async fn promote_scheduled(&self) { - if let Err(err) = self.promote_due_state("scheduled", "scheduled jobs").await { + #[tracing::instrument(skip(self, promoters), name = "maintenance.promote")] + async fn promote_scheduled(&self, promoters: &LeaderPromoters) { + if let Err(err) = self + .promote_due_state(JobState::Scheduled, "scheduled jobs", promoters) + .await + { error!(error = %err, "Failed to promote scheduled jobs"); } if let Err(err) = self - .promote_due_state("retryable", "retryable jobs (backoff elapsed)") + .promote_due_state( + JobState::Retryable, + "retryable jobs (backoff elapsed)", + promoters, + ) .await { error!(error = %err, "Failed to promote retryable jobs"); @@ -2556,138 +2821,62 @@ impl MaintenanceService { async fn promote_due_state( &self, - state: &'static str, + state: JobState, label: &'static str, - ) -> Result<(), awa_model::AwaError> { + promoters: &LeaderPromoters, + ) -> Result<(), AwaError> { + let canonical_state = match state { + JobState::Scheduled => "scheduled", + JobState::Retryable => "retryable", + other => { + return Err(AwaError::Validation(format!( + "unsupported promote state: {other:?}" + ))); + } + }; let mut promoted_total = 0usize; let mut notified_queues = HashSet::new(); - for _ in 0..PROMOTE_MAX_BATCHES_PER_TICK { - if self.cancel.is_cancelled() { - break; - } - - match &self.storage { - RuntimeStorage::Canonical => { - let (promoted, queues) = self - .promote_due_batch(state) - .await - .map_err(awa_model::AwaError::Database)?; - if promoted == 0 { - break; - } - - promoted_total += promoted; - notified_queues.extend(queues); - - if promoted < PROMOTE_BATCH_SIZE as usize { - break; - } - } - RuntimeStorage::QueueStorage(runtime) => { - let job_state = match state { - "scheduled" => awa_model::JobState::Scheduled, - "retryable" => awa_model::JobState::Retryable, - other => { - return Err(awa_model::AwaError::Validation(format!( - "unsupported queue storage promote state: {other}" - ))); - } - }; - let promote_start = std::time::Instant::now(); - let promoted = runtime - .store - .promote_due(&self.pool, job_state, PROMOTE_BATCH_SIZE) - .await?; - self.metrics.record_promotion_batch( - state, - promoted as u64, - promote_start.elapsed(), - ); - if promoted == 0 { - break; - } - - promoted_total += promoted; - - if promoted < PROMOTE_BATCH_SIZE as usize { - break; + match &self.storage { + RuntimeStorage::Canonical => { + promoted_total += self + .promote_canonical_due(canonical_state, &mut notified_queues) + .await?; + + // #456 mirror: a canonical-resolved (pre-flip) leader must + // promote the queue-storage deferred backlog once routing has + // flipped — re-scheduled canonical jobs migrate there, and + // post-flip runtimes that would promote them may not hold + // leadership. The active-schema guard keeps this a no-op until + // the flip. + if let (Some(runtime), Some(promoter)) = + (&self.standby_queue_storage, &promoters.standby) + { + let active_schema: Option = + sqlx::query_scalar("SELECT awa.active_queue_storage_schema()") + .fetch_one(&self.pool) + .await + .map_err(AwaError::Database)?; + if active_schema.as_deref() == Some(runtime.store.schema()) { + promoted_total += promoter.promote_due(state, &self.cancel).await?; } } } - } - - // #456: while a storage transition drains, due jobs can still sit in - // the canonical deferred backlog. Drain-only runtimes execute - // canonical work but stop leading maintenance once a post-flip - // runtime wins the election, so a queue-storage leader must promote - // the canonical plane too. Both queries are cheap no-ops once the - // canonical tables are empty (every finalized or fresh install). - if matches!(&self.storage, RuntimeStorage::QueueStorage(_)) { - for _ in 0..PROMOTE_MAX_BATCHES_PER_TICK { - if self.cancel.is_cancelled() { - break; - } - let (promoted, queues) = self - .promote_due_batch(state) - .await - .map_err(awa_model::AwaError::Database)?; - if promoted == 0 { - break; + RuntimeStorage::QueueStorage(_) => { + if let Some(promoter) = &promoters.primary { + promoted_total += promoter.promote_due(state, &self.cancel).await?; } - promoted_total += promoted; - notified_queues.extend(queues); - if promoted < PROMOTE_BATCH_SIZE as usize { - break; - } - } - } - // #456 mirror: a canonical-resolved (pre-flip) leader must promote - // the queue-storage deferred backlog once routing has flipped — - // re-scheduled canonical jobs migrate there, and post-flip runtimes - // that would promote them may not hold leadership. The active-schema - // guard keeps this a no-op until the flip. - if matches!(&self.storage, RuntimeStorage::Canonical) { - if let Some(runtime) = &self.standby_queue_storage { - let active_schema: Option = - sqlx::query_scalar("SELECT awa.active_queue_storage_schema()") - .fetch_one(&self.pool) - .await - .map_err(awa_model::AwaError::Database)?; - if active_schema.as_deref() == Some(runtime.store.schema()) { - let job_state = match state { - "scheduled" => awa_model::JobState::Scheduled, - "retryable" => awa_model::JobState::Retryable, - other => { - return Err(awa_model::AwaError::Validation(format!( - "unsupported queue storage promote state: {other}" - ))); - } - }; - for _ in 0..PROMOTE_MAX_BATCHES_PER_TICK { - if self.cancel.is_cancelled() { - break; - } - let promote_start = std::time::Instant::now(); - let promoted = runtime - .store - .promote_due(&self.pool, job_state, PROMOTE_BATCH_SIZE) - .await?; - self.metrics.record_promotion_batch( - state, - promoted as u64, - promote_start.elapsed(), - ); - if promoted == 0 { - break; - } - promoted_total += promoted; - if promoted < PROMOTE_BATCH_SIZE as usize { - break; - } - } - } + // #456: while a storage transition drains, due jobs can still + // sit in the canonical deferred backlog. Drain-only runtimes + // execute canonical work but stop leading maintenance once a + // post-flip runtime wins the election, so a queue-storage + // leader must promote the canonical plane too. This is a + // cheap no-op once the canonical tables are empty (every + // finalized or fresh install). + promoted_total += self + .promote_canonical_due(canonical_state, &mut notified_queues) + .await?; } } @@ -2695,7 +2884,7 @@ impl MaintenanceService { debug!( count = promoted_total, queues = notified_queues.len(), - state, + state = canonical_state, "Promoted {label}" ); } @@ -2703,6 +2892,34 @@ impl MaintenanceService { Ok(()) } + /// Promote due canonical `scheduled_jobs` rows in batches until one comes + /// back short. + async fn promote_canonical_due( + &self, + state: &'static str, + notified_queues: &mut HashSet, + ) -> Result { + let mut promoted_total = 0usize; + for _ in 0..PROMOTE_MAX_BATCHES_PER_TICK { + if self.cancel.is_cancelled() { + break; + } + let (promoted, queues) = self + .promote_due_batch(state) + .await + .map_err(AwaError::Database)?; + if promoted == 0 { + break; + } + promoted_total += promoted; + notified_queues.extend(queues); + if promoted < PROMOTE_BATCH_SIZE as usize { + break; + } + } + Ok(promoted_total) + } + /// SQL template for promotion. The state literal is injected directly /// (not as a parameter) so the planner can match the partial index on /// `(run_at, id) WHERE state = ''`. With a parameter, the planner @@ -3715,6 +3932,15 @@ mod tests { } } + #[test] + fn lane_waiters_scale_with_the_pool_and_never_drop_to_zero() { + assert_eq!(lane_waiter_limit(1), 1); + assert_eq!(lane_waiter_limit(10), 1); + assert_eq!(lane_waiter_limit(16), 2); + assert_eq!(lane_waiter_limit(32), 4); + assert_eq!(lane_waiter_limit(200), 4); + } + #[test] fn compute_fire_times_coalesces_missed_existing_fires_by_default() { let last = Utc.with_ymd_and_hms(2026, 5, 7, 12, 0, 0).unwrap(); diff --git a/awa/Cargo.toml b/awa/Cargo.toml index b928a5a4e..04bb5dee0 100644 --- a/awa/Cargo.toml +++ b/awa/Cargo.toml @@ -25,6 +25,7 @@ sqlx.workspace = true serde.workspace = true serde_json.workspace = true tokio = { workspace = true, features = ["full", "test-util"] } +tokio-util.workspace = true async-trait.workspace = true chrono.workspace = true proptest = "1" diff --git a/awa/tests/enqueue_lane_maintenance_coupling_test.rs b/awa/tests/enqueue_lane_maintenance_coupling_test.rs new file mode 100644 index 000000000..ce7d53040 --- /dev/null +++ b/awa/tests/enqueue_lane_maintenance_coupling_test.rs @@ -0,0 +1,427 @@ +//! How a saturated enqueue lane affects the maintenance leader. +//! +//! Promotion of scheduled and retryable jobs re-enqueues them into their lane, +//! so it queues on the same per-lane advisory lock as producers. The leader +//! runs its branches (promotion, rescue, rotation, ...) one after another on a +//! single task, so a promotion parked behind a hot lane would hold up the due +//! jobs of other queues and every other maintenance branch. The leader's +//! promoter gives each lane its own short transaction and hands a lane whose +//! lock stays busy to a dedicated waiter, so neither should wait on it. +//! +//! `test_maintenance_under_hot_lane` measures, with and without concurrent +//! producers on one hot lane: +//! * how long each promotion tick holds the leader task, and how long a +//! backlog on the hot queue takes to drain, +//! * promotion lag on an unrelated queue (time from a job becoming due to +//! it being visible as ready), with the promotion loop shaped like the +//! leader's (a tick, then batches until one comes back short), +//! * heartbeat rescue of abandoned claims on the hot queue: how long each run +//! takes and how late it starts once its tick is due (1s here, so a +//! blocked promotion shows up as a late start). +//! +//! Tunables (optional): +//! AWA_LANE_MAINT_PRODUCERS producers on the hot lane, default 16 +//! AWA_LANE_MAINT_BACKLOG due scheduled jobs on the hot queue, default 24576 +//! AWA_LANE_MAINT_RESCUE abandoned claims on the hot queue, default 6000 +//! +//! Requires a running Postgres instance. +//! Run with: +//! cargo test --release --package awa --test enqueue_lane_maintenance_coupling_test \ +//! -- --ignored --nocapture --test-threads=1 + +use awa::model::{insert_many, QueueStorage}; +use awa::worker::maintenance::QueueStoragePromoter; +use awa::worker::AwaMetrics; +use awa::{InsertOpts, InsertParams, JobState, QueueStorageConfig}; +use awa_testing::setup; +use sqlx::PgPool; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::time::{Duration, Instant}; +use uuid::Uuid; + +const PROMOTE_TICK: Duration = Duration::from_millis(250); +const PROBE_PERIOD: Duration = Duration::from_millis(100); +const RESCUE_TICK: Duration = Duration::from_secs(1); + +fn env_usize(name: &str, default: usize) -> usize { + std::env::var(name) + .ok() + .and_then(|raw| raw.parse().ok()) + .unwrap_or(default) +} + +fn percentile(values: &[f64], quantile: f64) -> f64 { + if values.is_empty() { + return 0.0; + } + let mut sorted = values.to_vec(); + sorted.sort_by(|left, right| left.partial_cmp(right).expect("finite")); + sorted[((sorted.len() as f64 - 1.0) * quantile).round() as usize] +} + +fn params( + queue: &str, + tag: &str, + n: u64, + run_at: Option>, +) -> InsertParams { + InsertParams { + kind: "lane_maint_job".to_string(), + args: serde_json::json!({ "tag": tag, "n": n }), + opts: InsertOpts { + queue: queue.to_string(), + run_at, + ..Default::default() + }, + } +} + +async fn seed_due_backlog(pool: &PgPool, queue: &str, count: usize) { + let due = chrono::Utc::now() - chrono::Duration::seconds(60); + let mut inserted = 0usize; + while inserted < count { + let chunk = (count - inserted).min(2_000); + let jobs: Vec = (0..chunk) + .map(|offset| params(queue, "backlog", (inserted + offset) as u64, Some(due))) + .collect(); + insert_many(pool, &jobs).await.expect("seed backlog"); + inserted += chunk; + } +} + +async fn deferred_due(pool: &PgPool, queue: &str) -> i64 { + sqlx::query_scalar( + "SELECT count(*)::bigint FROM awa.deferred_jobs WHERE queue = $1 AND state = 'scheduled'", + ) + .bind(queue) + .fetch_one(pool) + .await + .expect("count deferred") +} + +struct Producers { + stop: Arc, + tasks: Vec>, +} + +impl Producers { + fn start(pool: &PgPool, queue: &str, count: usize) -> Self { + let stop = Arc::new(AtomicBool::new(false)); + let tag = Uuid::new_v4().simple().to_string(); + let tasks = (0..count) + .map(|producer| { + let pool = pool.clone(); + let stop = Arc::clone(&stop); + let queue = queue.to_string(); + let tag = tag.clone(); + tokio::spawn(async move { + let mut jobs_inserted = 0u64; + let mut counter = (producer as u64) << 40; + while !stop.load(Ordering::Relaxed) { + let jobs: Vec = (0..185) + .map(|offset| params(&queue, &tag, counter + offset, None)) + .collect(); + counter += 185; + insert_many(&pool, &jobs).await.expect("producer insert"); + jobs_inserted += 185; + } + jobs_inserted + }) + }) + .collect(); + Self { stop, tasks } + } + + async fn stop(self) -> u64 { + self.stop.store(true, Ordering::Relaxed); + let mut total = 0; + for task in self.tasks { + total += task.await.expect("producer task"); + } + total + } +} + +#[derive(Default)] +struct LeaderReport { + /// How long each promotion tick held the leader task. + tick_ms: Vec, + drain_s: f64, + probe_lag_ms: Vec, + /// How long after its tick each heartbeat-rescue run started, and how long + /// the run itself took. + rescue_start_delay_ms: Vec, + rescue_run_ms: Vec, +} + +/// A leader loop shaped like the maintenance leader's: one task selecting over +/// a promotion timer (250ms; each tick runs the leader's promoter) and a +/// heartbeat-rescue timer, running each branch to completion before taking +/// the next tick. Runs until the hot queue's backlog is gone and +/// `min_window` has passed, while a probe feeds one due job per 100ms into an +/// unrelated queue and times how long each takes to become ready. +async fn run_leader( + pool: &PgPool, + store: &Arc, + hot: &str, + other: &str, + min_window: Duration, +) -> LeaderReport { + let mut report = LeaderReport::default(); + let started = Instant::now(); + let stop = Arc::new(AtomicBool::new(false)); + let promoter = + QueueStoragePromoter::new(pool.clone(), Arc::clone(store), AwaMetrics::from_global()); + let cancel = tokio_util::sync::CancellationToken::new(); + + let probe = { + let pool = pool.clone(); + let other = other.to_string(); + let stop = Arc::clone(&stop); + tokio::spawn(async move { + let mut lags = Vec::new(); + let mut n = 0u64; + while !stop.load(Ordering::Relaxed) { + n += 1; + // Older than the hot backlog, so the probe is always in the + // first promotion batch rather than behind the backlog's + // run_at order. + let due_at = chrono::Utc::now() - chrono::Duration::seconds(120); + insert_many(&pool, &[params(&other, "probe", n, Some(due_at))]) + .await + .expect("insert probe"); + let inserted = Instant::now(); + loop { + let visible: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM awa.ready_entries + WHERE queue = $1 AND args->>'tag' = 'probe' AND (args->>'n')::bigint = $2)", + ) + .bind(&other) + .bind(n as i64) + .fetch_one(&pool) + .await + .expect("probe visibility"); + if visible { + lags.push(inserted.elapsed().as_secs_f64() * 1e3); + break; + } + if stop.load(Ordering::Relaxed) { + return lags; + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + tokio::time::sleep(PROBE_PERIOD).await; + } + lags + }) + }; + + let mut promote_timer = tokio::time::interval(PROMOTE_TICK); + promote_timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + let mut rescue_timer = tokio::time::interval(RESCUE_TICK); + rescue_timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + promote_timer.tick().await; + rescue_timer.tick().await; + + let mut backlog_gone_at: Option = None; + loop { + tokio::select! { + tick = promote_timer.tick() => { + let _ = tick; + let tick_started = Instant::now(); + promoter + .promote_due(JobState::Scheduled, &cancel) + .await + .expect("promote_due"); + report.tick_ms.push(tick_started.elapsed().as_secs_f64() * 1e3); + if backlog_gone_at.is_none() && deferred_due(pool, hot).await == 0 { + backlog_gone_at = Some(Instant::now()); + } + if backlog_gone_at.is_some() && started.elapsed() >= min_window { + break; + } + } + tick = rescue_timer.tick() => { + let delay = tokio::time::Instant::now().saturating_duration_since(tick); + let run_started = Instant::now(); + let rescued = store + .rescue_stale_heartbeats(pool, Duration::ZERO) + .await + .expect("rescue_stale_heartbeats"); + if !rescued.is_empty() { + report.rescue_start_delay_ms.push(delay.as_secs_f64() * 1e3); + report.rescue_run_ms.push(run_started.elapsed().as_secs_f64() * 1e3); + } + } + } + } + report.drain_s = backlog_gone_at + .expect("backlog drained") + .duration_since(started) + .as_secs_f64(); + stop.store(true, Ordering::Relaxed); + report.probe_lag_ms = probe.await.expect("probe task"); + report +} + +fn max_of(values: &[f64]) -> f64 { + values.iter().copied().fold(0.0, f64::max) +} + +fn print_leader(label: &str, report: &LeaderReport) { + println!( + "[lane-maint] {label} promote tick on leader ms p50={:.0} p99={:.0} max={:.0} | hot backlog \ + drained in {:.2}s | unrelated-queue promotion lag ms p50={:.0} p99={:.0} max={:.0} ({} probes) \ + | rescue ran {} times, run ms p50={:.0} max={:.0}, started late by ms p50={:.0} max={:.0}", + percentile(&report.tick_ms, 0.5), + percentile(&report.tick_ms, 0.99), + max_of(&report.tick_ms), + report.drain_s, + percentile(&report.probe_lag_ms, 0.5), + percentile(&report.probe_lag_ms, 0.99), + max_of(&report.probe_lag_ms), + report.probe_lag_ms.len(), + report.rescue_run_ms.len(), + percentile(&report.rescue_run_ms, 0.5), + max_of(&report.rescue_run_ms), + percentile(&report.rescue_start_delay_ms, 0.5), + max_of(&report.rescue_start_delay_ms), + ); +} + +/// Enqueue `count` jobs on `queue` and claim every one of them without ever +/// completing or heartbeating, so each becomes an abandoned attempt. +async fn seed_abandoned_claims(pool: &PgPool, store: &QueueStorage, queue: &str, count: usize) { + let tag = Uuid::new_v4().simple().to_string(); + let mut enqueued = 0usize; + while enqueued < count { + let chunk = (count - enqueued).min(2_000); + let jobs: Vec = (0..chunk) + .map(|offset| params(queue, &tag, (enqueued + offset) as u64, None)) + .collect(); + store + .enqueue_params_batch(pool, &jobs) + .await + .expect("enqueue claimable jobs"); + enqueued += chunk; + } + let mut claimed = 0usize; + while claimed < count { + let batch = store + .claim_runtime_batch(pool, queue, 500, Duration::from_secs(3_600)) + .await + .expect("claim batch"); + assert!(!batch.is_empty(), "ran out of jobs to claim"); + claimed += batch.len(); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 8)] +#[ignore] +async fn test_maintenance_under_hot_lane() { + let producers = env_usize("AWA_LANE_MAINT_PRODUCERS", 16); + let backlog = env_usize("AWA_LANE_MAINT_BACKLOG", 24_576); + let rescue_claims = env_usize("AWA_LANE_MAINT_RESCUE", 6_000); + + let pool = setup::setup(producers as u32 + 12).await; + setup::activate_queue_storage(&pool).await; + let store = Arc::new(QueueStorage::new(QueueStorageConfig::default()).expect("queue storage")); + + for (label, producer_count) in [("idle lane ", 0usize), ("hot lane ", producers)] { + let suffix = Uuid::new_v4().simple().to_string(); + let hot = format!("lane_maint_hot_{}", &suffix[..10]); + let other = format!("lane_maint_other_{}", &suffix[..10]); + + // Warm both lanes so lane provisioning is not measured. + insert_many( + &pool, + &[ + params(&hot, "warm", 0, None), + params(&other, "warm", 0, None), + ], + ) + .await + .expect("warm lanes"); + + seed_due_backlog(&pool, &hot, backlog).await; + seed_abandoned_claims(&pool, &store, &hot, rescue_claims).await; + + let running = (producer_count > 0).then(|| Producers::start(&pool, &hot, producer_count)); + tokio::time::sleep(Duration::from_millis(1_500)).await; + + let report = run_leader(&pool, &store, &hot, &other, Duration::from_secs(12)).await; + print_leader(label, &report); + + if let Some(running) = running { + let produced = running.stop().await; + println!("[lane-maint] {label} producers inserted {produced} jobs meanwhile"); + } + } +} + +/// The mechanism without producer load: one transaction holds the hot lane's +/// reservation lock for a fixed time while the leader loop runs. Promotion of +/// a batch that contains hot-queue rows waits for it, and takes the unrelated +/// queue's due job in the same batch and the leader's other branches with it. +/// Heartbeat rescue called directly does not wait for the lane at all: it +/// writes retryable rows to the deferred table, which promotion later moves. +#[tokio::test(flavor = "multi_thread", worker_threads = 8)] +#[ignore] +async fn test_leader_behind_a_held_lane_lock() { + let hold = Duration::from_millis(env_usize("AWA_LANE_MAINT_HOLD_MS", 3_000) as u64); + let pool = setup::setup(16).await; + setup::activate_queue_storage(&pool).await; + let store = Arc::new(QueueStorage::new(QueueStorageConfig::default()).expect("queue storage")); + + let suffix = Uuid::new_v4().simple().to_string(); + let hot = format!("lane_maint_hot_{}", &suffix[..10]); + let other = format!("lane_maint_other_{}", &suffix[..10]); + insert_many( + &pool, + &[ + params(&hot, "warm", 0, None), + params(&other, "warm", 0, None), + ], + ) + .await + .expect("warm lanes"); + seed_due_backlog(&pool, &hot, 8_192).await; + seed_abandoned_claims(&pool, &store, &hot, 3_000).await; + + let key: i64 = sqlx::query_scalar( + "SELECT hashtextextended( + format('%I.%I', 'awa', awa.queue_lane_sequence_name('queue_enqueue_seq', $1, 2::smallint, 0::smallint)), + 0)", + ) + .bind(&hot) + .fetch_one(&pool) + .await + .expect("derive lane lock key"); + let mut gate = pool.begin().await.expect("begin gate"); + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(key) + .execute(&mut *gate) + .await + .expect("take lane lock"); + + let rescue_started = Instant::now(); + let rescued = store + .rescue_stale_heartbeats(&pool, Duration::ZERO) + .await + .expect("rescue with the lane lock held"); + println!( + "[lane-maint] rescue of {} claims with the hot lane lock held took {:.0}ms", + rescued.len(), + rescue_started.elapsed().as_secs_f64() * 1e3 + ); + let remaining_hold = hold.saturating_sub(rescue_started.elapsed()); + + let release = tokio::spawn(async move { + tokio::time::sleep(remaining_hold).await; + gate.commit().await.expect("release lane lock"); + }); + let report = run_leader(&pool, &store, &hot, &other, Duration::from_secs(9)).await; + release.await.expect("release task"); + print_leader(&format!("lane held {}ms", hold.as_millis()), &report); +} diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index c693f4e30..a58a8bdf1 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -6,12 +6,13 @@ mod ci_timing; #[path = "../../awa-model/src/storage_footprint_sql.rs"] mod storage_footprint_sql; +use ci_timing::scaled_timeout; use awa::audited_sql; use awa::model::{ admin, batch_operations, insert, migrations, storage, AwaError, BatchOperationFilter, - BatchOperationSpec, PruneOutcome, QueueStorage, QueueStorageConfig, RingRotationCadence, - RotateOutcome, SkipReason, SubmitBatchOperation, + BatchOperationSpec, PromotionLane, PruneOutcome, QueueStorage, QueueStorageConfig, + RingRotationCadence, RotateOutcome, SkipReason, SubmitBatchOperation, }; use awa::{ Client, InsertOpts, JobArgs, JobContext, JobError, JobResult, JobRow, JobState, QueueConfig, @@ -11108,9 +11109,10 @@ fn branch_overruns() -> Arc>> { /// snoozed and retry-after jobs runnable, so overrunning promotion bodies must /// never arm a cooldown on it. /// -/// The first three promotion bodies (two `DELETE`s each) take 1.2s against a -/// 25ms promote interval, so the maintenance branch tracker flags promotion as -/// overrunning. The test asserts on that overrun warning rather than on job +/// Every promotion body runs the canonical-plane promotion `DELETE` once per +/// state, even with nothing due. A statement trigger slows the first six, so +/// the first three bodies take 1.2s against a 25ms promote interval and the +/// maintenance branch tracker flags promotion as overrunning. The test asserts on that overrun warning rather than on job /// latency: the promotion warning must report no cooldown, while another /// branch in the same run (cleanup, whose bodies exceed its 1ms interval) /// does report one, so the check cannot pass vacuously. Jobs that fall due @@ -11137,7 +11139,7 @@ async fn test_queue_storage_runtime_promotion_not_suppressed_after_overrunning_b END $$" ), format!( - "CREATE TRIGGER slow_first_promote_deletes BEFORE DELETE ON {schema}.deferred_jobs \ + "CREATE TRIGGER slow_first_promote_deletes BEFORE DELETE ON awa.scheduled_jobs \ FOR EACH STATEMENT EXECUTE FUNCTION {schema}.slow_first_promote_deletes()" ), ] { @@ -11272,6 +11274,668 @@ async fn test_queue_storage_runtime_promotion_not_suppressed_after_overrunning_b client.shutdown(Duration::from_secs(5)).await; } +/// Advisory lock key of a lane's enqueue reservation, as +/// `reserve_enqueue_seq` takes it. The lane must already exist. +async fn enqueue_lane_lock_key( + pool: &sqlx::PgPool, + store: &QueueStorage, + queue: &str, + priority: i16, +) -> i64 { + sqlx::query_scalar(audited_sql(format!( + "SELECT hashtextextended(format('%I.%I', $1::text, seq_name), 0) \ + FROM {}.queue_enqueue_heads \ + WHERE queue = $2 AND priority = $3 AND enqueue_shard = 0", + store.schema() + ))) + .bind(store.schema()) + .bind(queue) + .bind(priority) + .fetch_one(pool) + .await + .expect("derive enqueue lane lock key") +} + +/// A promotion batch that meets a lane whose enqueue lock is held promotes +/// the other lanes, reports the held lane as contended after its lock timeout +/// and leaves its rows deferred. Promoted later, the lane's rows take lane +/// sequence numbers in due order and are all claimable. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_queue_storage_promotion_leaves_a_contended_lane_deferred_and_in_order() { + const PRIORITY: i16 = 2; + let (_db_guard, pool) = setup_pool(6).await; + let hot = "qs_promote_contended_hot"; + let other = "qs_promote_contended_other"; + let store = create_store(&pool, "awa_qs_promote_contended_lane").await; + let opts = |queue: &str, run_at: Option>| InsertOpts { + queue: queue.to_string(), + priority: PRIORITY, + run_at, + ..Default::default() + }; + + let warm_job_id = enqueue_job(&pool, &store, &CompleteJob { id: 0 }, opts(hot, None)).await; + // Later job ids fall due earlier, so due order differs from id order. + let due_base = Utc::now() - chrono::Duration::seconds(60); + let mut hot_due = Vec::new(); + for n in 0..6 { + let run_at = due_base - chrono::Duration::seconds(n); + let job_id = enqueue_job( + &pool, + &store, + &CompleteJob { id: n }, + opts(hot, Some(run_at)), + ) + .await; + hot_due.push((run_at, job_id)); + } + hot_due.sort(); + let hot_in_due_order: Vec = hot_due.iter().map(|(_, job_id)| *job_id).collect(); + let other_job_id = enqueue_job( + &pool, + &store, + &CompleteJob { id: 100 }, + opts(other, Some(due_base)), + ) + .await; + + let hot_lane = PromotionLane { + queue: hot.to_string(), + priority: PRIORITY, + }; + let lock_key = enqueue_lane_lock_key(&pool, &store, hot, PRIORITY).await; + let mut holder = pool.begin().await.expect("begin lane holder"); + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(lock_key) + .execute(holder.as_mut()) + .await + .expect("hold the hot lane's enqueue lock"); + + let lock_timeout = Duration::from_millis(100); + let started = Instant::now(); + let outcome = store + .promote_due_lanes(&pool, JobState::Scheduled, 100, Some(lock_timeout), &[]) + .await + .expect("promote with the hot lane held"); + let elapsed = started.elapsed(); + assert_eq!(outcome.promoted, 1, "{outcome:?}"); + assert_eq!(outcome.selected, 7, "{outcome:?}"); + assert_eq!(outcome.contended, vec![hot_lane.clone()]); + assert!( + elapsed >= lock_timeout && elapsed < scaled_timeout(Duration::from_secs(1)), + "the held lane should cost one lock timeout, took {elapsed:?}" + ); + let other_job = store + .load_job(&pool, other_job_id) + .await + .expect("load other job") + .expect("other job exists"); + assert_eq!(other_job.state, JobState::Available); + for job_id in &hot_in_due_order { + let job = store + .load_job(&pool, *job_id) + .await + .expect("load hot job") + .expect("hot job exists"); + assert_eq!(job.state, JobState::Scheduled, "job {job_id}"); + } + + let skipped = store + .promote_due_lanes( + &pool, + JobState::Scheduled, + 100, + Some(lock_timeout), + std::slice::from_ref(&hot_lane), + ) + .await + .expect("promote with the hot lane skipped"); + assert_eq!(skipped, Default::default()); + + holder.commit().await.expect("release the hot lane"); + let mut batches = 0; + loop { + let outcome = store + .promote_due_lane(&pool, JobState::Scheduled, &hot_lane, 4, None) + .await + .expect("promote the hot lane"); + assert!(outcome.contended.is_empty(), "{outcome:?}"); + if outcome.promoted == 0 { + break; + } + batches += 1; + } + assert_eq!(batches, 2); + + let in_lane_order: Vec = sqlx::query_scalar(audited_sql(format!( + "SELECT job_id FROM {}.ready_entries WHERE job_id = ANY($1) ORDER BY lane_seq", + store.schema() + ))) + .bind(&hot_in_due_order) + .fetch_all(&pool) + .await + .expect("read hot lane order"); + assert_eq!(in_lane_order, hot_in_due_order); + + let mut claimed: Vec = store + .claim_runtime_batch(&pool, hot, 100, Duration::from_secs(60)) + .await + .expect("claim the hot lane") + .into_iter() + .map(|claimed| claimed.job.id) + .collect(); + claimed.sort_unstable(); + let mut expected = hot_in_due_order.clone(); + expected.push(warm_job_id); + expected.sort_unstable(); + assert_eq!(claimed, expected); +} + +/// A physical promotion lane has one waiter for both deferred states. Its +/// batches preserve each state's due order and publish lane sequence prefixes. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_queue_storage_promotion_shares_one_waiter_between_states() { + use opentelemetry::metrics::MeterProvider as _; + + const BATCH_SIZE: usize = 4_096; + const SCHEDULED: usize = BATCH_SIZE + 3; + const PRIORITY: i16 = 2; + let (_db_guard, pool) = setup_pool(16).await; + let hot = "qs_mixed_state_hot"; + let other = "qs_mixed_state_other"; + let store = Arc::new(create_store(&pool, "awa_qs_mixed_state_waiter").await); + let opts = |queue: &str, run_at: Option>| InsertOpts { + queue: queue.to_string(), + priority: PRIORITY, + run_at, + ..Default::default() + }; + let warm = enqueue_job(&pool, &store, &CompleteJob { id: -1 }, opts(hot, None)).await; + let due_base = Utc::now() - chrono::Duration::seconds(60); + let scheduled: Vec<_> = (0..SCHEDULED) + .map(|index| { + insert::params_with( + &CompleteJob { id: index as i64 }, + opts( + hot, + Some(due_base - chrono::Duration::milliseconds(index as i64)), + ), + ) + .expect("scheduled params") + }) + .collect(); + for chunk in scheduled.chunks(2_000) { + store + .enqueue_params_batch(&pool, chunk) + .await + .expect("seed scheduled rows"); + } + let scheduled_order: Vec = sqlx::query_scalar(audited_sql(format!( + "SELECT job_id FROM {}.deferred_jobs WHERE queue = $1 ORDER BY run_at,job_id", + store.schema() + ))) + .bind(hot) + .fetch_all(&pool) + .await + .expect("scheduled due order"); + assert_eq!(scheduled_order.len(), SCHEDULED); + let mut retryable = Vec::new(); + for index in 0..3 { + retryable.push( + enqueue_job( + &pool, + &store, + &CompleteJob { id: -(index + 2) }, + opts(hot, Some(due_base - chrono::Duration::seconds(index + 1))), + ) + .await, + ); + } + sqlx::query(audited_sql(format!( + "UPDATE {}.deferred_jobs SET state = 'retryable' WHERE job_id = ANY($1)", + store.schema() + ))) + .bind(&retryable) + .execute(&pool) + .await + .expect("seed retryable state"); + let retryable_order: Vec = sqlx::query_scalar(audited_sql(format!( + "SELECT job_id FROM {}.deferred_jobs WHERE job_id = ANY($1) ORDER BY run_at,job_id", + store.schema() + ))) + .bind(&retryable) + .fetch_all(&pool) + .await + .expect("retryable due order"); + let unrelated = enqueue_job( + &pool, + &store, + &CompleteJob { id: -9 }, + opts(other, Some(due_base)), + ) + .await; + let shards: i64 = sqlx::query_scalar(audited_sql(format!( + "SELECT count(*) FROM {}.queue_enqueue_heads WHERE queue = $1 AND priority = $2", + store.schema() + ))) + .bind(hot) + .bind(PRIORITY) + .fetch_one(&pool) + .await + .expect("count hot enqueue shards"); + assert_eq!(shards, 1, "the two states must share one physical lock"); + + let exporter = InMemoryMetricExporter::default(); + let meter_provider = SdkMeterProvider::builder() + .with_periodic_exporter(exporter.clone()) + .build(); + let metrics = awa::worker::AwaMetrics::new(&meter_provider.meter("mixed-state-promotion")); + let promoter = awa::worker::maintenance::QueueStoragePromoter::new( + pool.clone(), + Arc::clone(&store), + metrics, + ); + let cancel = tokio_util::sync::CancellationToken::new(); + let lock_key = enqueue_lane_lock_key(&pool, &store, hot, PRIORITY).await; + let mut holder = pool.begin().await.expect("begin lane holder"); + let holder_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(holder.as_mut()) + .await + .expect("holder pid"); + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(lock_key) + .execute(holder.as_mut()) + .await + .expect("hold the hot lane"); + assert_eq!( + promoter + .promote_due(JobState::Scheduled, &cancel) + .await + .expect("scheduled tick"), + 1 + ); + assert_eq!( + promoter + .promote_due(JobState::Retryable, &cancel) + .await + .expect("retryable tick"), + 0 + ); + assert_eq!( + store + .load_job(&pool, unrelated) + .await + .expect("unrelated job") + .expect("unrelated exists") + .state, + JobState::Available + ); + meter_provider.force_flush().expect("flush hand-offs"); + assert_eq!( + sum_counter_metric_with_attribute( + &exporter.get_finished_metrics().expect("exported metrics"), + "awa.maintenance.promote.lane_contention", + "awa.promote.outcome", + "handed_off" + ), + 1, + "scheduled and retryable must share one waiter" + ); + let deadline = Instant::now() + scaled_timeout(Duration::from_secs(2)); + loop { + let waiters: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_stat_activity WHERE datname = current_database() + AND wait_event_type = 'Lock' AND wait_event = 'advisory' + AND $1 = ANY(pg_blocking_pids(pid))", + ) + .bind(holder_pid) + .fetch_one(&pool) + .await + .expect("lane waiters"); + if waiters > 0 { + assert_eq!(waiters, 1); + break; + } + assert!(Instant::now() < deadline, "dedicated waiter never queued"); + tokio::time::sleep(Duration::from_millis(10)).await; + } + holder.commit().await.expect("release the hot lane"); + + let mut expected = vec![warm]; + expected.extend_from_slice(&scheduled_order[..BATCH_SIZE]); + expected.extend_from_slice(&retryable_order); + expected.extend_from_slice(&scheduled_order[BATCH_SIZE..]); + let deadline = Instant::now() + scaled_timeout(Duration::from_secs(10)); + let published = loop { + let rows: Vec<(i64, i64)> = sqlx::query_as(audited_sql(format!( + "SELECT job_id,lane_seq FROM {}.ready_entries WHERE queue = $1 ORDER BY lane_seq", + store.schema() + ))) + .bind(hot) + .fetch_all(&pool) + .await + .expect("published lane prefix"); + let ids: Vec<_> = rows.iter().map(|(id, _)| *id).collect(); + assert!( + expected.starts_with(&ids), + "later lane work committed ahead of its prefix" + ); + assert!( + rows.windows(2).all(|rows| rows[0].1 < rows[1].1), + "lane sequences must advance" + ); + if ids.len() == expected.len() { + break ids; + } + assert!( + Instant::now() < deadline, + "both deferred states did not drain" + ); + tokio::time::sleep(Duration::from_millis(10)).await; + }; + let scheduled_ids: HashSet<_> = scheduled_order.iter().copied().collect(); + let retryable_ids: HashSet<_> = retryable_order.iter().copied().collect(); + assert_eq!( + published + .iter() + .copied() + .filter(|id| scheduled_ids.contains(id)) + .collect::>(), + scheduled_order + ); + assert_eq!( + published + .iter() + .copied() + .filter(|id| retryable_ids.contains(id)) + .collect::>(), + retryable_order + ); + let remaining: i64 = sqlx::query_scalar(audited_sql(format!( + "SELECT count(*) FROM {}.deferred_jobs WHERE queue = $1", + store.schema() + ))) + .bind(hot) + .fetch_one(&pool) + .await + .expect("remaining hot rows"); + assert_eq!(remaining, 0); + drop(promoter); +} + +/// The maintenance leader must not wait on a lane whose enqueue lock never +/// comes free. Two sessions take turns holding the hot lane's lock for +/// `HOLD` each, so a lock request queued behind them waits up to two holds. +/// A due job on another queue must still run, and an abandoned attempt must +/// still be rescued, well within one hold; the hot queue's due jobs must +/// still be promoted while the lane stays busy. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_queue_storage_runtime_busy_lane_holds_back_neither_promotion_nor_rescue() { + const HOLD: Duration = Duration::from_secs(5); + const PRIORITY: i16 = 2; + let (_db_guard, pool) = setup_pool(12).await; + let hot = "qs_busy_lane_hot"; + let other = "qs_busy_lane_other"; + let abandoned = "qs_busy_lane_abandoned"; + let schema = "awa_qs_busy_lane_leader"; + let store = create_store(&pool, schema).await; + let opts = |queue: &str, run_at: Option>| InsertOpts { + queue: queue.to_string(), + priority: PRIORITY, + run_at, + ..Default::default() + }; + + enqueue_job(&pool, &store, &CompleteJob { id: 0 }, opts(hot, None)).await; + let lock_key = enqueue_lane_lock_key(&pool, &store, hot, PRIORITY).await; + + let stop = Arc::new(AtomicBool::new(false)); + let mut holders = Vec::new(); + for _ in 0..2 { + let (held_tx, held_rx) = tokio::sync::oneshot::channel::<()>(); + let pool = pool.clone(); + let stop = Arc::clone(&stop); + holders.push(tokio::spawn(async move { + let mut held_tx = Some(held_tx); + let mut holds = 0u32; + while !stop.load(Ordering::SeqCst) { + let mut tx = pool.begin().await.expect("begin lane holder"); + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(lock_key) + .execute(tx.as_mut()) + .await + .expect("hold the hot lane's enqueue lock"); + holds += 1; + if let Some(held_tx) = held_tx.take() { + let _ = held_tx.send(()); + } + tokio::time::sleep(HOLD).await; + tx.commit().await.expect("release the hot lane"); + } + holds + })); + // The first holder takes the lock; wait for it so the second queues + // behind it rather than racing it. + if holders.len() == 1 { + held_rx.await.expect("first holder took the lane"); + } + } + + let due = Utc::now() - chrono::Duration::seconds(1); + let mut hot_jobs = Vec::new(); + for n in 1..=3 { + hot_jobs + .push(enqueue_job(&pool, &store, &CompleteJob { id: n }, opts(hot, Some(due))).await); + } + let other_job = enqueue_job( + &pool, + &store, + &CompleteJob { id: 10 }, + opts(other, Some(due)), + ) + .await; + let abandoned_job = enqueue_job( + &pool, + &store, + &CompleteJob { id: 20 }, + opts(abandoned, None), + ) + .await; + let claimed = store + .claim_runtime_batch(&pool, abandoned, 1, Duration::from_secs(3_600)) + .await + .expect("claim the abandoned attempt"); + assert_eq!(claimed.len(), 1); + + let heartbeat_staleness = Duration::from_millis(500); + let worker_queue = || QueueConfig { + max_workers: 4, + poll_interval: Duration::from_millis(25), + ..QueueConfig::default() + }; + let client = Client::builder(pool.clone()) + .queue(hot, worker_queue()) + .queue(other, worker_queue()) + .queue_storage( + QueueStorageConfig { + schema: schema.to_string(), + queue_slot_count: 4, + lease_slot_count: 2, + lease_claim_receipts: false, + ..Default::default() + }, + Duration::from_secs(60), + Duration::from_secs(60), + ) + .claim_rotate_interval(Duration::from_secs(60)) + .register_worker(CompleteWorker) + .promote_interval(Duration::from_millis(50)) + .leader_election_interval(Duration::from_millis(100)) + .leader_check_interval(Duration::from_millis(50)) + .heartbeat_rescue_interval(Duration::from_millis(100)) + .heartbeat_staleness(heartbeat_staleness) + .deadline_rescue_interval(Duration::from_secs(10)) + .callback_rescue_interval(Duration::from_secs(10)) + .build() + .expect("Failed to build busy-lane client"); + client + .start() + .await + .expect("Failed to start busy-lane client"); + + let bound = scaled_timeout(Duration::from_secs(2)); + wait_for_job_state(&store, &pool, other_job, &[JobState::Completed], bound).await; + wait_for_job_state( + &store, + &pool, + abandoned_job, + &[JobState::Retryable, JobState::Available], + bound + heartbeat_staleness, + ) + .await; + + for job_id in &hot_jobs { + wait_for_job_state( + &store, + &pool, + *job_id, + &[JobState::Completed], + scaled_timeout(HOLD * 3), + ) + .await; + } + assert!( + holders.iter().all(|holder| !holder.is_finished()), + "the hot lane must stay busy until its jobs are promoted" + ); + + stop.store(true, Ordering::SeqCst); + for holder in holders { + assert!(holder.await.expect("lane holder") >= 1); + } + client.shutdown(Duration::from_secs(5)).await; +} + +/// Rows another transaction holds locked, as a lane promotion still waiting +/// under a former leader does, must not fill every batch of a promotion tick. +/// With more of the oldest due rows locked than one batch selects, a due job +/// on another queue must still be promoted and run promptly, and the locked +/// rows must be promoted once released. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_queue_storage_runtime_locked_due_rows_do_not_starve_promotion() { + const LOCKED: usize = 4_096 + 64; + let (_db_guard, pool) = setup_pool(10).await; + let locked_queue = "qs_locked_due_rows"; + let other = "qs_locked_due_other"; + let schema = "awa_qs_locked_due_rows"; + let store = create_store(&pool, schema).await; + + let oldest = Utc::now() - chrono::Duration::seconds(120); + let params: Vec<_> = (0..LOCKED) + .map(|n| { + insert::params_with( + &CompleteJob { id: n as i64 }, + InsertOpts { + queue: locked_queue.to_string(), + run_at: Some(oldest), + ..Default::default() + }, + ) + .expect("build locked-row params") + }) + .collect(); + for chunk in params.chunks(2_000) { + store + .enqueue_params_batch(&pool, chunk) + .await + .expect("enqueue rows to lock"); + } + let mut holder = pool.begin().await.expect("begin row holder"); + let locked: Vec = sqlx::query_scalar(audited_sql(format!( + "SELECT job_id FROM {schema}.deferred_jobs WHERE queue = $1 FOR UPDATE" + ))) + .bind(locked_queue) + .fetch_all(holder.as_mut()) + .await + .expect("lock the oldest due rows"); + assert_eq!(locked.len(), LOCKED); + + let other_job = enqueue_job( + &pool, + &store, + &CompleteJob { id: -1 }, + InsertOpts { + queue: other.to_string(), + run_at: Some(Utc::now() - chrono::Duration::seconds(1)), + ..Default::default() + }, + ) + .await; + + let client = Client::builder(pool.clone()) + .queue( + other, + QueueConfig { + max_workers: 4, + poll_interval: Duration::from_millis(25), + ..QueueConfig::default() + }, + ) + .queue_storage( + QueueStorageConfig { + schema: schema.to_string(), + queue_slot_count: 4, + lease_slot_count: 2, + lease_claim_receipts: false, + ..Default::default() + }, + Duration::from_secs(60), + Duration::from_secs(60), + ) + .claim_rotate_interval(Duration::from_secs(60)) + .register_worker(CompleteWorker) + .promote_interval(Duration::from_millis(50)) + .leader_election_interval(Duration::from_millis(100)) + .leader_check_interval(Duration::from_millis(50)) + .build() + .expect("Failed to build locked-rows client"); + client + .start() + .await + .expect("Failed to start locked-rows client"); + + wait_for_job_state( + &store, + &pool, + other_job, + &[JobState::Completed], + scaled_timeout(Duration::from_secs(2)), + ) + .await; + + holder.rollback().await.expect("release the locked rows"); + let deadline = Instant::now() + scaled_timeout(Duration::from_secs(10)); + loop { + let remaining: i64 = sqlx::query_scalar(audited_sql(format!( + "SELECT count(*) FROM {schema}.deferred_jobs WHERE queue = $1" + ))) + .bind(locked_queue) + .fetch_one(&pool) + .await + .expect("count deferred rows"); + if remaining == 0 { + break; + } + assert!( + Instant::now() < deadline, + "{remaining} released rows were never promoted" + ); + tokio::time::sleep(Duration::from_millis(50)).await; + } + + client.shutdown(Duration::from_secs(5)).await; +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn test_queue_storage_runtime_stale_heartbeat_rescue() { let (_db_guard, pool) = setup_pool(10).await; diff --git a/docs/architecture.md b/docs/architecture.md index d3c527f61..af066bf37 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -147,7 +147,7 @@ sequenceDiagram Enqueue is transactional: if the producer's outer transaction rolls back, the job never becomes visible. Immediate jobs reserve a lane sequence range under a transaction-scoped lane lock, append to `ready_entries_*`, and append a compact `ready_segments` range for the committed lane sequence range; future scheduled or retryable jobs append to `deferred_jobs`. The lane lock is held until commit or rollback so a later producer cannot make lane `N+1` visible while lane `N` can still commit. -`deferred_jobs` is deliberately outside the claim path. The maintenance leader promotes due `scheduled` and `retryable` rows into `ready_entries_*` in batches and notifies the target queues. Cron schedules are just producers for ordinary jobs: when a schedule fires, the cron transaction records the fire and enqueues the job atomically, using `ready_entries_*` for immediate fires or `deferred_jobs` when the enqueue carries a future `run_at`. +`deferred_jobs` is deliberately outside the claim path. The maintenance leader promotes due `scheduled` and `retryable` rows into `ready_entries_*` in batches and notifies the target queues. Re-enqueueing a row takes its lane lock, which producers hold until they commit, so each `(queue, priority)` lane's share of a batch is promoted in its own transaction under a short `lock_timeout`. A lane that times out is handed to one dedicated waiter task shared by scheduled and retryable rows; the leader skips both states for that lane. The waiter alternates states within a total 32-batch budget while other lanes and the leader's rescue, rotation and prune continue. Each lock acquisition has a 30 s timeout. The combined wait behind several finite producer transactions can exhaust it, releasing the waiter's queue position; later ticks retry, but sustained contention can still prevent promotion. Cron schedules are just producers for ordinary jobs: when a schedule fires, the cron transaction records the fire and enqueues the job atomically, using `ready_entries_*` for immediate fires or `deferred_jobs` when the enqueue carries a future `run_at`. There are two COPY-shaped producer paths: diff --git a/docs/configuration.md b/docs/configuration.md index f4e36389f..7c79abce3 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -245,6 +245,7 @@ Retry backoff and `Snooze` use the same mechanism: the current attempt closes, a Operational knobs: - `promote_interval` controls how often maintenance checks for due deferred rows. +- A queue whose enqueue lane is kept busy by producers does not delay promotion of other queues or the leader's rescue: promotion waits 50 ms for a lane's lock, then hands the lane to one waiter shared by its scheduled and retryable rows. The leader skips that lane in both states while its waiter alternates states within a total 32-batch budget. Each waiter holds one connection from the worker's pool while it waits, on top of the leader's own. The leader runs at most one waiter per eight pool connections, never fewer than one or more than four, so a 10-connection pool gives up at most one extra connection to a busy lane. `awa_maintenance_promote_lane_contention_total` counts those hand-offs by queue. A steady `waiter_limit` rate means more lanes are saturated than the waiters cover, and a larger pool allows more waiters. `waiter_timed_out` means a lock acquisition exhausted its 30 s wait budget, including time queued behind several finite transactions. That timeout releases the waiter's queue position; a later tick retries, so sustained contention can still prevent promotion. - Per-job `run_at` controls the due time. - Periodic schedules are declared in worker code with `periodic()` and managed through the CLI/UI `cron` surface. diff --git a/docs/grafana/README.md b/docs/grafana/README.md index 68c0e0ae6..f1d031b28 100644 --- a/docs/grafana/README.md +++ b/docs/grafana/README.md @@ -254,6 +254,7 @@ In Prometheus they appear under the OpenTelemetry [OTLP-to-Prometheus translatio | `awa.maintenance.promote_batches` | Counter | state | Promotion batches | | `awa.maintenance.promote_batch_size` | Histogram | state | Jobs promoted | | `awa.maintenance.promote_duration` | Histogram (s) | state | Promotion time | +| `awa.maintenance.promote.lane_contention` | Counter | state, queue, outcome | Promotion lanes whose lock was busy: `handed_off` to a dedicated waiter, `waiter_limit` when every waiter was busy, `waiter_timed_out` when a waiter's 30 s wait ran out | | `awa.maintenance.rescues` | Counter | rescue_kind | Jobs rescued | | `awa.maintenance.prune.attempts` | Counter | ring, outcome, reason | Ring prune outcomes, including destructive `pruned` and no-DDL `already_pruned` | | `awa.maintenance.prune.duration` | Histogram (s) | ring, phase | Successful destructive prune time split into `lock`, `truncate`, and `commit` | diff --git a/skills/awa-operations/SKILL.md b/skills/awa-operations/SKILL.md index a3f2cea37..b0354291d 100644 --- a/skills/awa-operations/SKILL.md +++ b/skills/awa-operations/SKILL.md @@ -140,7 +140,16 @@ tick). The leader alone runs cluster-wide scans: deferred→ready promotion, the three rescue paths (stale-heartbeat, hard-deadline, callback-timeout), ring rotation and prune, DLQ and descriptor cleanup, and cron evaluation. During an unfinalized transition it promotes both planes' deferred backlogs, so drain and -post-flip execution do not depend on which runtime holds leadership. Every +post-flip execution do not depend on which runtime holds leadership. +Promotion gives each queue lane its own short transaction and hands a lane +whose enqueue lock stays busy to one waiter shared by scheduled and retryable +rows. The waiter alternates states within a total 32-batch budget while the +leader skips that lane in both states and continues other promotion and rescue. +A 30 s lock wait can expire behind several finite producer transactions; the +waiter releases its queue position and a later tick retries. Sustained +contention can still prevent promotion. +`awa_maintenance_promote_lane_contention_total` counts the hand-offs and timed-out +waits. Every worker dispatches and heartbeats its own attempts. There is no `pg_cron` or external scheduler.