diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5dd4872b2..7cc80e01e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -108,6 +108,7 @@ jobs: ports: - 5432:5432 options: >- + --shm-size=256m --health-cmd pg_isready --health-interval 10s --health-timeout 5s @@ -153,6 +154,7 @@ jobs: ports: - 5432:5432 options: >- + --shm-size=256m --health-cmd pg_isready --health-interval 10s --health-timeout 5s @@ -222,6 +224,7 @@ jobs: ports: - 5432:5432 options: >- + --shm-size=256m --health-cmd pg_isready --health-interval 10s --health-timeout 5s @@ -262,6 +265,7 @@ jobs: ports: - 5432:5432 options: >- + --shm-size=256m --health-cmd pg_isready --health-interval 10s --health-timeout 5s @@ -295,6 +299,7 @@ jobs: ports: - 5432:5432 options: >- + --shm-size=256m --health-cmd pg_isready --health-interval 10s --health-timeout 5s @@ -335,6 +340,7 @@ jobs: ports: - 5432:5432 options: >- + --shm-size=256m --health-cmd pg_isready --health-interval 10s --health-timeout 5s @@ -417,6 +423,7 @@ jobs: ports: - 5432:5432 options: >- + --shm-size=256m --health-cmd pg_isready --health-interval 10s --health-timeout 5s @@ -488,6 +495,7 @@ jobs: ports: - 5432:5432 options: >- + --shm-size=256m --health-cmd pg_isready --health-interval 10s --health-timeout 5s @@ -586,6 +594,7 @@ jobs: ports: - 5432:5432 options: >- + --shm-size=256m --health-cmd pg_isready --health-interval 10s --health-timeout 5s diff --git a/CHANGELOG.md b/CHANGELOG.md index 244d516b3..c256abad8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,9 @@ Notable changes between releases. Detailed migration notes for storage transitio ### Fixed - Admin job lists and kind counts work on a queue-storage schema installed with more claim slots than the default, instead of failing with `invalid physical claim slot`. +- Heartbeat, deadline and callback rescue of lease-backed attempts reach the stale attempts behind a full batch of attempts whose retained ready bodies are ambiguous. Before, every pass selected the same ambiguous attempts first, left them running, and rescued nothing else. A pass now sets all of a batch's ambiguous attempts aside at once. +- Admin job lists and kind counts no longer read other claim slots' receipt closure partitions when they read one slot's open claims. +- A `Completed` lifecycle event, and the job a `Completed` follow-up enqueue spec receives, no longer carry the finished attempt's `heartbeat_at` and `deadline_at`; both are `None`, as a read of the completed job returns them. ## [0.6.14] — 2026-11-08 diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index 90a774b0b..6fae9ae27 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -865,20 +865,20 @@ fn open_receipt_running_claims_sql_from( WHERE {batch_filter}NOT ( batches.receipt_ranges <@ COALESCE(( SELECT range_agg(closed) - FROM {schema}.lease_claim_closure_batches AS closure_batches + FROM {closure_batch_rel} AS closure_batches CROSS JOIN LATERAL unnest(closure_batches.receipt_ranges) AS closed WHERE closure_batches.claim_slot = batches.claim_slot AND closure_batches.receipt_ranges && batches.receipt_ranges ), '{{}}'::int8multirange) ) AND NOT EXISTS ( - SELECT 1 FROM {schema}.lease_claim_closures AS closures + SELECT 1 FROM {closure_rel} AS closures WHERE closures.claim_slot = batches.claim_slot AND closures.job_id = items.job_id AND closures.run_lease = items.run_lease ) AND NOT EXISTS ( - SELECT 1 FROM {schema}.lease_claim_closure_batches AS closure_batches + SELECT 1 FROM {closure_batch_rel} AS closure_batches WHERE closure_batches.claim_slot = batches.claim_slot AND closure_batches.receipt_ranges @> items.receipt_id -- Keep the range lookup correlated: joining on claim_slot can @@ -1756,25 +1756,49 @@ struct ReadyRowIdentity { } /// Attempts a rescue pass leaves for a later pass, as `(job_id, run_lease)`. -struct BusyRetained(Vec<(i64, i64)>); +struct BusyRetained { + attempts: Vec<(i64, i64)>, + /// The attempts stay unresolvable until their retained bodies are + /// repaired, rather than until a lock is released. + unresolvable: bool, +} impl BusyRetained { - fn exclude(self, excluded: &mut BTreeSet<(i64, i64)>) -> Result<(), AwaError> { + fn busy(attempts: Vec<(i64, i64)>) -> Self { + Self { + attempts, + unresolvable: false, + } + } + + /// Leaves the attempts out of the rest of the pass. Busy attempts shrink + /// the pass's selection. Unresolvable attempts are replaced in it, so a + /// backlog of them, which every pass meets first, cannot fill each batch + /// and keep rescue from the attempts behind it. + fn exclude( + self, + excluded: &mut BTreeSet<(i64, i64)>, + selected: &mut Option>, + ) -> Result<(), AwaError> { let previous = excluded.len(); - excluded.extend(self.0); + excluded.extend(self.attempts); if excluded.len() == previous { return Err(AwaError::Validation( "queue storage retained attempt retry made no progress".to_string(), )); } + if self.unresolvable { + *selected = None; + } Ok(()) } } -/// A failure hydrating a deleted lease. `Ambiguous` names an attempt whose -/// retained ready bodies rank equally, so none can be chosen for it. +/// A failure hydrating deleted leases. `Ambiguous` names the attempts, as +/// `(job_id, run_lease)`, whose retained ready bodies rank equally, so none +/// can be chosen for them. enum HydrateError { - Ambiguous { job_id: i64, run_lease: i64 }, + Ambiguous(Vec<(i64, i64)>), Awa(AwaError), } @@ -1787,9 +1811,16 @@ impl From for HydrateError { impl From for AwaError { fn from(err: HydrateError) -> Self { match err { - HydrateError::Ambiguous { job_id, run_lease } => AwaError::Validation(format!( - "queue storage ready body ambiguous for deleted lease job {job_id} run_lease {run_lease}" - )), + HydrateError::Ambiguous(attempts) => AwaError::Validation(match attempts.as_slice() { + [] => "queue storage ready body ambiguous for deleted lease".to_string(), + [(job_id, run_lease)] => format!( + "queue storage ready body ambiguous for deleted lease job {job_id} run_lease {run_lease}" + ), + [(job_id, run_lease), rest @ ..] => format!( + "queue storage ready body ambiguous for deleted lease job {job_id} run_lease {run_lease} and {} more", + rest.len() + ), + }), HydrateError::Awa(err) => err, } } @@ -2456,10 +2487,10 @@ fn select_claimed_ready_body<'a>( let mut best = candidates.iter().filter(|row| rank(row) == strongest); let ready = best.next().expect("ready body rank has a candidate"); if best.next().is_some() { - return Err(HydrateError::Ambiguous { - job_id: reference.job_id, - run_lease: reference.run_lease, - }); + return Err(HydrateError::Ambiguous(vec![( + reference.job_id, + reference.run_lease, + )])); } Ok((ready, has_emitted)) } @@ -10849,14 +10880,19 @@ impl QueueStorage { match hydrated { Ok(hydrated) => Ok(hydrated), // One unresolvable attempt must not stop the rescue of the others. - Err(HydrateError::Ambiguous { job_id, run_lease }) => { - tracing::error!( - schema = self.schema(), - job_id, - run_lease, - "Rescue skipped an attempt: queue storage ready body ambiguous for deleted lease; it stays running until its retained bodies are repaired" - ); - Ok(Err(BusyRetained(vec![(job_id, run_lease)]))) + Err(HydrateError::Ambiguous(attempts)) => { + for (job_id, run_lease) in &attempts { + tracing::error!( + schema = self.schema(), + job_id, + run_lease, + "Rescue skipped an attempt: queue storage ready body ambiguous for deleted lease; it stays running until its retained bodies are repaired" + ); + } + Ok(Err(BusyRetained { + attempts, + unresolvable: true, + })) } Err(HydrateError::Awa(err)) => Err(err), } @@ -10876,6 +10912,33 @@ impl QueueStorage { let run_leases: Vec = deleted.iter().map(|row| row.run_lease).collect(); let (ready_map, claimed_lanes) = self.load_ready_snapshot_candidates_tx(tx, &deleted).await?; + // Report every ambiguous attempt at once, before any write, so a + // caller that sets them aside needs one retry for all of them. + let ambiguous: Vec<(i64, i64)> = deleted + .iter() + .filter(|row| { + ready_map + .get(&( + row.ready_slot, + row.ready_generation, + row.queue.clone(), + row.enqueue_shard, + row.lane_seq, + row.job_id, + row.run_lease, + )) + .is_some_and(|candidates| { + matches!( + select_claimed_ready_body(candidates, &claimed_lanes, row), + Err(HydrateError::Ambiguous(_)) + ) + }) + }) + .map(|row| (row.job_id, row.run_lease)) + .collect(); + if !ambiguous.is_empty() { + return Err(HydrateError::Ambiguous(ambiguous)); + } let attempt_rows: Vec = sqlx::query_as(&format!( r#" @@ -11130,10 +11193,10 @@ impl QueueStorage { .insert((source.job_id, source.run_lease), source) .is_some() { - return Err(HydrateError::Ambiguous { - job_id: source.job_id, - run_lease: source.run_lease + 1, - }); + return Err(HydrateError::Ambiguous(vec![( + source.job_id, + source.run_lease + 1, + )])); } } let job_ids: Vec = sources.iter().map(|row| row.job_id).collect(); @@ -11181,7 +11244,7 @@ impl QueueStorage { .await .map_err(map_sqlx_error)?; if !busy.is_empty() { - return Ok(Some(BusyRetained(busy))); + return Ok(Some(BusyRetained::busy(busy))); } } else { self.lock_claim_heads_of_jobs_with_wait_tx(tx, &job_ids, true) @@ -11266,10 +11329,10 @@ impl QueueStorage { for row in &emitted { let key = (row.job_id, row.run_lease); if !seen.insert(key) || expected.get(&key).copied() != Some(row) { - return Err(HydrateError::Ambiguous { - job_id: row.job_id, - run_lease: row.run_lease + 1, - }); + return Err(HydrateError::Ambiguous(vec![( + row.job_id, + row.run_lease + 1, + )])); } } sqlx::query(&format!( @@ -15408,7 +15471,7 @@ impl QueueStorage { Ok(moved) => moved, Err(busy) => { attempt.rollback().await.map_err(map_sqlx_error)?; - busy.exclude(&mut excluded)?; + busy.exclude(&mut excluded, &mut selected)?; continue; } }; @@ -15419,7 +15482,7 @@ impl QueueStorage { Ok(moved) => moved, Err(busy) => { attempt.rollback().await.map_err(map_sqlx_error)?; - busy.exclude(&mut excluded)?; + busy.exclude(&mut excluded, &mut selected)?; continue; } }; @@ -15578,7 +15641,7 @@ impl QueueStorage { Ok(moved) => moved, Err(busy) => { attempt.rollback().await.map_err(map_sqlx_error)?; - busy.exclude(&mut excluded)?; + busy.exclude(&mut excluded, &mut selected)?; continue; } }; @@ -15590,7 +15653,7 @@ impl QueueStorage { Ok(moved) => moved, Err(busy) => { attempt.rollback().await.map_err(map_sqlx_error)?; - busy.exclude(&mut excluded)?; + busy.exclude(&mut excluded, &mut selected)?; continue; } }, @@ -15716,7 +15779,7 @@ impl QueueStorage { Ok(moved) => moved, Err(busy) => { attempt.rollback().await.map_err(map_sqlx_error)?; - busy.exclude(&mut excluded)?; + busy.exclude(&mut excluded, &mut selected)?; continue; } }; diff --git a/awa-worker/src/executor.rs b/awa-worker/src/executor.rs index f68955775..d1c5db277 100644 --- a/awa-worker/src/executor.rs +++ b/awa-worker/src/executor.rs @@ -1723,6 +1723,17 @@ fn rescheduled_event_row( row } +/// The job as a read returns it after this attempt completes. +fn completed_event_row(job: &JobRow) -> JobRow { + let mut row = job.clone(); + row.state = JobState::Completed; + row.finalized_at = Some(chrono::Utc::now()); + row.progress = None; + row.heartbeat_at = None; + row.deadline_at = None; + row +} + #[allow(clippy::too_many_arguments)] async fn complete_job_queue_storage( runtime: &QueueStorageRuntime, @@ -1833,15 +1844,11 @@ async fn complete_job_queue_storage( return Ok(CompletionOutcome::IgnoredStale); } if needs_event { - // A completed attempt changes only these fields, so the event is - // built from the worker's copy rather than re-reading the job. - let mut updated_job = job.clone(); - updated_job.state = JobState::Completed; - updated_job.finalized_at = Some(chrono::Utc::now()); - updated_job.progress = None; + // The event is built from the worker's copy rather than + // re-reading the job. Ok(CompletionOutcome::Applied { event: Some(UntypedJobEvent::Completed { - job: updated_job, + job: completed_event_row(job), duration, }), terminal: false, @@ -2459,12 +2466,7 @@ async fn complete_queue_storage_with_followups( // Synthesise the post-completion snapshot — the store has just moved the // lease into `done_entries` so a live SELECT would race with rotation. - // The happy-path fields are exactly those `complete_job_queue_storage` - // reconstructs when `load_job` returns None. - let mut updated_job = job.clone(); - updated_job.state = JobState::Completed; - updated_job.finalized_at = Some(chrono::Utc::now()); - updated_job.progress = None; + let updated_job = completed_event_row(job); crate::enqueue_specs::dispatch_specs_in_tx(&mut tx, &updated_job, specs, None).await?; diff --git a/awa/tests/lifecycle_hook_test.rs b/awa/tests/lifecycle_hook_test.rs index 3eb72d9ff..1b413df12 100644 --- a/awa/tests/lifecycle_hook_test.rs +++ b/awa/tests/lifecycle_hook_test.rs @@ -9,7 +9,7 @@ use awa::{ }; use serde::{Deserialize, Serialize}; use sqlx::postgres::PgPoolOptions; -use std::sync::{Arc, OnceLock}; +use std::sync::{Arc, Mutex, OnceLock}; use std::time::Duration; use tokio::sync::{mpsc, Semaphore}; @@ -121,6 +121,11 @@ struct HookJob { value: String, } +#[derive(Debug, Clone, Serialize, Deserialize, JobArgs)] +struct HookFollowUp { + source_job_id: i64, +} + #[derive(Debug, Clone, Serialize, Deserialize, JobArgs)] struct RawHookJob { value: String, @@ -183,6 +188,104 @@ async fn test_typed_completed_event_handler_runs() { assert_eq!(stored.state, JobState::Completed); } +/// A completed event carries the job as a terminal read returns it, without the +/// finished attempt's heartbeat or deadline. +#[tokio::test] +async fn test_completed_event_of_a_deadline_job_matches_the_terminal_job() { + assert_completed_event_of_a_deadline_job_matches_the_terminal_job(false).await; +} + +/// A kind with a Completed follow-up completes through its own transaction; the +/// follow-up and the event see the same terminal job. +#[tokio::test] +async fn test_completed_event_with_a_follow_up_matches_the_terminal_job() { + assert_completed_event_of_a_deadline_job_matches_the_terminal_job(true).await; +} + +type AttemptTimes = ( + Option>, + Option>, +); + +async fn assert_completed_event_of_a_deadline_job_matches_the_terminal_job(with_follow_up: bool) { + let _permit = test_gate() + .acquire_owned() + .await + .expect("test gate should be available"); + let pool = setup_pool().await; + let queue = "lifecycle_completed_deadline"; + clean_queue(&pool, queue).await; + + let (tx, mut rx) = mpsc::unbounded_channel(); + let follow_up_saw: Arc>> = Arc::default(); + let builder = Client::builder(pool.clone()) + .queue( + queue, + QueueConfig { + poll_interval: Duration::from_millis(25), + deadline_duration: Duration::from_secs(60), + ..Default::default() + }, + ) + .register::(|_args, _ctx| async move { Ok(JobResult::Completed) }) + .on_event::(move |event| { + let tx = tx.clone(); + async move { + if let JobEvent::Completed { job, .. } = event { + tx.send(job).unwrap(); + } + } + }); + let builder = if with_follow_up { + let follow_up_saw = follow_up_saw.clone(); + builder.on_completed_enqueue::(move |_args, job| { + follow_up_saw + .lock() + .unwrap() + .push((job.heartbeat_at, job.deadline_at)); + HookFollowUp { + source_job_id: job.id, + } + }) + } else { + builder + }; + let client = builder.build().unwrap(); + + let inserted = awa::insert_with( + &pool, + &HookJob { + action: "complete".into(), + value: "deadline".into(), + }, + awa::InsertOpts { + queue: queue.to_string(), + ..Default::default() + }, + ) + .await + .unwrap(); + + client.start().await.unwrap(); + let event_job = recv_event(&mut rx).await; + client.shutdown(Duration::from_secs(2)).await; + + let stored = admin::get_job(&pool, inserted.id).await.unwrap(); + assert_eq!(event_job.id, stored.id); + assert_eq!(event_job.state, stored.state); + assert_eq!( + (event_job.heartbeat_at, event_job.deadline_at), + (stored.heartbeat_at, stored.deadline_at) + ); + assert_eq!((stored.heartbeat_at, stored.deadline_at), (None, None)); + let expected_follow_up_saw: Vec = if with_follow_up { + vec![(None, None)] + } else { + Vec::new() + }; + assert_eq!(*follow_up_saw.lock().unwrap(), expected_follow_up_saw); +} + #[tokio::test] async fn test_typed_started_event_handler_runs() { let _permit = test_gate() diff --git a/awa/tests/persona_soak/db.rs b/awa/tests/persona_soak/db.rs index d195a6c1a..a2ab41c8e 100644 --- a/awa/tests/persona_soak/db.rs +++ b/awa/tests/persona_soak/db.rs @@ -172,12 +172,17 @@ impl Budget { } } - /// Reserves up to `want` connections under `label`, never fewer than `min`. - /// Returns the amount granted; panics when even `min` does not fit, because the - /// run would exhaust the server before it tested anything. + /// Reserves up to `want` connections under `label`, never fewer than `min`, + /// replacing any earlier reservation under the same label. Returns the amount + /// granted; panics when even `min` does not fit, because the run would exhaust + /// the server before it tested anything. pub fn reserve(&self, label: &str, want: u32, min: u32) -> u32 { let mut allocations = self.allocations.lock().expect("budget lock"); - let used: u32 = allocations.values().sum(); + let used: u32 = allocations + .iter() + .filter(|(held_by, _)| held_by.as_str() != label) + .map(|(_, count)| count) + .sum(); let available = self.usable.saturating_sub(used); assert!( available >= min, @@ -216,7 +221,8 @@ pub type ClientFactory = Arc Client + Send + Sync>; /// shutdown handshake and the connections drop. pub struct ClientHost { pub label: String, - runtime: tokio::runtime::Runtime, + /// Always `Some` until the host is stopped or dropped. + runtime: Option, client: Arc, pool: PgPool, } @@ -250,7 +256,7 @@ impl ClientHost { })?; Ok(Self { label: thread_label, - runtime, + runtime: Some(runtime), client: Arc::new(client), pool, }) @@ -268,26 +274,95 @@ impl ClientHost { } /// Graceful stop: stop claiming, drain in-flight jobs, then close the pool. - pub async fn shutdown(self, timeout: Duration) { + pub async fn shutdown(mut self, timeout: Duration) { + let Some(runtime) = self.runtime.take() else { + return; + }; let client = self.client.clone(); let pool = self.pool.clone(); - let _ = self - .runtime + let _ = runtime .handle() .spawn(async move { client.shutdown(timeout).await; pool.close().await; }) .await; - self.runtime.shutdown_background(); + runtime.shutdown_background(); } /// Abrupt stop: no shutdown handshake, as when a process is killed. pub fn kill(self) { - self.runtime.shutdown_background(); + drop(self); + } +} + +/// Dropping a host kills it. A tokio runtime dropped the ordinary way blocks +/// until its tasks finish, which panics inside another runtime. +impl Drop for ClientHost { + fn drop(&mut self) { + if let Some(runtime) = self.runtime.take() { + runtime.shutdown_background(); + } } } pub fn url_for_database(config: &SoakConfig, name: &str) -> String { replace_database_name(&config.database_url, name) } + +#[cfg(test)] +mod tests { + use super::{Budget, ClientHost, ServerInfo}; + use awa::{Client, QueueConfig}; + use sqlx::postgres::PgPoolOptions; + use std::sync::Arc; + + fn budget(usable: u32) -> Budget { + Budget::new( + &ServerInfo { + max_connections: usable, + ..ServerInfo::default() + }, + 0, + ) + } + + #[test] + fn replacing_a_reservation_reuses_its_own_connections() { + let budget = budget(20); + budget.reserve("other", 10, 1); + assert_eq!(budget.reserve("earmark", 10, 2), 10); + assert_eq!( + budget.reserve("earmark", 10, 10), + 10, + "the earmark's own connections count as available to its replacement" + ); + assert_eq!( + budget.table(), + vec![("earmark".to_string(), 10), ("other".to_string(), 10)] + ); + } + + #[tokio::test] + async fn dropping_a_live_host_inside_another_runtime_does_not_panic() { + let pool = PgPoolOptions::new() + .connect_lazy("postgres://postgres:test@localhost:1/awa_soak_unused") + .expect("lazy pool"); + let client = Client::builder(pool.clone()) + .queue("awa_soak_host_drop", QueueConfig::default()) + .build() + .expect("build an unstarted client"); + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(1) + .enable_all() + .build() + .expect("host runtime"); + let host = ClientHost { + label: "host_drop".to_string(), + runtime: Some(runtime), + client: Arc::new(client), + pool, + }; + drop(host); + } +} diff --git a/awa/tests/persona_soak/oracle.rs b/awa/tests/persona_soak/oracle.rs index aa70b2e5b..ed5f53366 100644 --- a/awa/tests/persona_soak/oracle.rs +++ b/awa/tests/persona_soak/oracle.rs @@ -287,8 +287,7 @@ impl Oracle { let scan_started_ms = self.shared.now_ms(); let rows: Result, _> = sqlx::query_as(&format!( "SELECT job_id, queue, run_lease FROM {schema}.done_entries - WHERE state = 'cancelled' AND finalized_at > now() - make_interval(secs => $1) - LIMIT 2000" + WHERE state = 'cancelled' AND finalized_at > now() - make_interval(secs => $1)" )) .bind(((scan_started_ms - since_ms) as f64 / 1000.0 + 15.0).max(20.0)) .fetch_all(&self.pool) @@ -1175,6 +1174,46 @@ mod tests { } } + #[tokio::test] + async fn cancel_scan_reads_every_cancellation_in_its_window() { + let mut run = TestRun::create().await; + let schema = schema(&run.shared); + for (queue, count) in [("oracle_cancel_bulk", 2_000), ("oracle_cancel_late", 1)] { + run.store + .enqueue_batch(&run.pool, queue, 2, count) + .await + .expect("enqueue jobs to cancel"); + awa::model::admin::drain_queue(&run.pool, queue) + .await + .expect("cancel the queue"); + } + let late: i64 = sqlx::query_scalar(&format!( + "SELECT job_id FROM {schema}.done_entries WHERE queue = 'oracle_cancel_late'" + )) + .fetch_one(&run.pool) + .await + .expect("the late cancellation"); + run.shared.ledger.register( + late, + JobRecord::new(Persona::Priya, u32::MAX, 0, expect::COMPLETED, 0), + ); + run.oracle.check_unexplained_cancels().await; + assert!( + run.shared + .violations + .lock() + .expect("violations") + .iter() + .any(|violation| violation.kind == "unexplained_cancel" + && violation + .rows + .iter() + .any(|row| row.contains(&format!("\"job_id\":{late},")))), + "a cancellation after the first 2,000 in the window is still checked" + ); + run.drop().await; + } + #[tokio::test] async fn returned_handler_without_finalization_cannot_pass_the_oracle() { let mut run = TestRun::create().await; diff --git a/awa/tests/persona_soak/personas/alex.rs b/awa/tests/persona_soak/personas/alex.rs index 88498cf04..ccd3f1698 100644 --- a/awa/tests/persona_soak/personas/alex.rs +++ b/awa/tests/persona_soak/personas/alex.rs @@ -396,9 +396,10 @@ impl Context { } } 2 => { - // Bulk retry: every exhausted Mia job may come back. - self.shared.ledger.mark_where( - |record| record.persona == Persona::Mia && record.has(flag::EXHAUSTED), + // Bulk retry: every exhausted job on the infer queue may come back. + super::mia::mark_exhausted_in_queue( + &self.shared, + super::mia::INFER_QUEUE, flag::DLQ_RETRIED, ); let bulk = ListDlqFilter { @@ -416,8 +417,9 @@ impl Context { } } _ => { - self.shared.ledger.mark_where( - |record| record.persona == Persona::Mia && record.has(flag::EXHAUSTED), + super::mia::mark_exhausted_in_queue( + &self.shared, + super::mia::TRAIN_QUEUE, flag::DLQ_PURGED, ); let bulk = ListDlqFilter { diff --git a/awa/tests/persona_soak/personas/chaos.rs b/awa/tests/persona_soak/personas/chaos.rs index fee1c63eb..99def38ee 100644 --- a/awa/tests/persona_soak/personas/chaos.rs +++ b/awa/tests/persona_soak/personas/chaos.rs @@ -14,6 +14,23 @@ use std::time::{Duration, Instant}; const MAINTENANCE_LEADER_LOCK_KEY: i64 = 0x4157_415f_4d41_494e; const RUNTIME_FLEETS: [&str; 6] = ["mia", "tom", "priya", "sam", "sam-stripe", "fanout"]; +/// Up to `$1` random runtime sessions. Producers, operators, the oracle, chaos +/// itself, the claim-and-abandon persona and the shard-ops control sessions are +/// left alone: they are the harness, not the system under test. +macro_rules! runtime_victims_sql { + () => { + "SELECT pid FROM pg_stat_activity + WHERE application_name LIKE 'soak\\_%' AND application_name NOT LIKE '%producer%' + AND application_name NOT LIKE '%operator%' AND application_name NOT LIKE '%oracle%' + AND application_name NOT LIKE '%chaos%' AND application_name NOT LIKE 'soak\\_abandon\\_%' + AND application_name NOT IN + ('soak_shardops_owner', 'soak_shardops_migrator', 'soak_shardops_admin') + AND backend_type = 'client backend' + AND pid <> pg_backend_pid() + ORDER BY random() LIMIT $1" + }; +} + async fn kill_runtime(shared: &Arc, rng: &mut Rng) { let Some(&name) = rng.pick(&RUNTIME_FLEETS) else { return; @@ -92,17 +109,11 @@ async fn terminate_leader(shared: &Arc, pool: &PgPool) { async fn terminate_runtime_connections(shared: &Arc, pool: &PgPool, rng: &mut Rng) { let count = rng.range(3, 12) as i64; - let terminated: Result, _> = sqlx::query_as( - "SELECT pid FROM ( - SELECT pid FROM pg_stat_activity - WHERE application_name LIKE 'soak\\_%' AND application_name NOT LIKE '%producer%' - AND application_name NOT LIKE '%operator%' AND application_name NOT LIKE '%oracle%' - AND application_name NOT LIKE '%chaos%' AND application_name NOT LIKE 'soak\\_abandon\\_%' - AND backend_type = 'client backend' - AND pid <> pg_backend_pid() - ORDER BY random() LIMIT $1 - ) victims WHERE pg_terminate_backend(pid)", - ) + let terminated: Result, _> = sqlx::query_as(concat!( + "SELECT pid FROM (", + runtime_victims_sql!(), + ") victims WHERE pg_terminate_backend(pid)" + )) .bind(count) .fetch_all(pool) .await; @@ -240,3 +251,50 @@ pub async fn run(shared: Arc) -> Result<(), String> { pool.close().await; Ok(()) } + +#[cfg(test)] +mod tests { + use crate::persona_soak::db::connect_pool; + use std::collections::HashSet; + + #[tokio::test] + async fn runtime_termination_spares_the_shard_ops_control_sessions() { + let url = std::env::var("DATABASE_URL") + .unwrap_or_else(|_| "postgres://postgres:test@localhost:15432/awa_test".to_string()); + let mut sessions = Vec::new(); + for application in [ + "soak_shardops_owner", + "soak_shardops_migrator", + "soak_shardops_admin", + "soak_shardops-a_0", + ] { + let pool = connect_pool(&url, application, 1).await.expect("session"); + let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&pool) + .await + .expect("backend pid"); + sessions.push((application, pid, pool)); + } + let probe = connect_pool(&url, "chaos_victim_probe", 1) + .await + .expect("probe"); + let victims: HashSet = sqlx::query_scalar(runtime_victims_sql!()) + .bind(i64::from(i32::MAX)) + .fetch_all(&probe) + .await + .expect("select victims") + .into_iter() + .collect(); + for (application, pid, _) in &sessions { + assert_eq!( + victims.contains(pid), + *application == "soak_shardops-a_0", + "{application}" + ); + } + probe.close().await; + for (_, _, pool) in sessions { + pool.close().await; + } + } +} diff --git a/awa/tests/persona_soak/personas/mia.rs b/awa/tests/persona_soak/personas/mia.rs index 07d77f276..0100a2589 100644 --- a/awa/tests/persona_soak/personas/mia.rs +++ b/awa/tests/persona_soak/personas/mia.rs @@ -258,3 +258,58 @@ pub async fn run(shared: Arc) -> Result<(), String> { producer_pool.close().await; Ok(()) } + +/// Flags Mia's exhausted jobs on `queue`: the rows a DLQ operation filtered to that +/// queue acts on. Returns how many were flagged. +pub fn mark_exhausted_in_queue(shared: &Shared, queue: &str, flags: u32) -> usize { + let Some(queue) = shared.queues.get(queue) else { + return 0; + }; + shared.ledger.mark_where( + |record| { + record.persona == Persona::Mia + && record.queue == queue.index + && record.has(flag::EXHAUSTED) + }, + flags, + ) +} + +#[cfg(test)] +mod tests { + use super::{mark_exhausted_in_queue, DEADLINE_QUEUE, INFER_QUEUE, TRAIN_QUEUE}; + use crate::persona_soak::config::{Persona, SoakConfig}; + use crate::persona_soak::db::{Budget, ServerInfo}; + use crate::persona_soak::ledger::{expect, flag, JobRecord}; + use crate::persona_soak::shared::Shared; + + #[test] + fn a_dlq_operation_flags_only_its_own_queue() { + let config = SoakConfig::for_test( + "postgres://postgres:test@localhost/awa_soak_dlq_marks".to_string(), + ); + let info = ServerInfo::default(); + let budget = Budget::new(&info, 0); + let shared = Shared::new(config, info, budget); + for (id, queue) in [(1, TRAIN_QUEUE), (2, INFER_QUEUE), (3, DEADLINE_QUEUE)] { + let index = shared.queues.register(queue, Persona::Mia, false).index; + shared.ledger.register( + id, + JobRecord::new( + Persona::Mia, + index, + 0, + expect::ANY_TERMINAL, + flag::EXHAUSTED, + ), + ); + } + assert_eq!( + mark_exhausted_in_queue(&shared, TRAIN_QUEUE, flag::DLQ_PURGED), + 1 + ); + assert!(shared.ledger.get(1).unwrap().has(flag::DLQ_PURGED)); + assert!(!shared.ledger.get(2).unwrap().has(flag::DLQ_PURGED)); + assert!(!shared.ledger.get(3).unwrap().has(flag::DLQ_PURGED)); + } +} diff --git a/awa/tests/persona_soak/personas/python.rs b/awa/tests/persona_soak/personas/python.rs index 7ece76691..0433f0bba 100644 --- a/awa/tests/persona_soak/personas/python.rs +++ b/awa/tests/persona_soak/personas/python.rs @@ -141,6 +141,20 @@ fn spawn_process( .env("SOAK_QUEUE_WORK", QUEUE_WORK) .env("SOAK_QUEUE_DEADLINE", QUEUE_DEADLINE) .env("SOAK_QUEUE_SHARED", QUEUE_SHARED) + .env( + "SOAK_HEARTBEAT_MS", + shared.config.heartbeat_interval.as_millis().to_string(), + ) + .env( + "SOAK_STALENESS_MS", + shared.config.heartbeat_staleness.as_millis().to_string(), + ) + .envs( + shared + .config + .rescue_interval + .map(|interval| ("SOAK_RESCUE_INTERVAL_MS", interval.as_millis().to_string())), + ) .stdout(Stdio::piped()) .stderr(Stdio::piped()) .kill_on_drop(true) diff --git a/awa/tests/persona_soak/personas/tom.rs b/awa/tests/persona_soak/personas/tom.rs index af963e56c..3f3b5ede3 100644 --- a/awa/tests/persona_soak/personas/tom.rs +++ b/awa/tests/persona_soak/personas/tom.rs @@ -54,6 +54,12 @@ impl Worker for TomWorker { } } +/// Tenant queues to serve: the scaled count, and never fewer than the queues the cron +/// schedules write to. +fn queue_count(scaled: u64) -> usize { + (scaled as usize).max(CRON_JOBS) +} + fn queue_name(index: usize) -> String { format!("{PREFIX}{index:03}") } @@ -109,7 +115,7 @@ fn client_factory(shared: Arc, queue_count: usize) -> ClientFactory { pub async fn run(shared: Arc) -> Result<(), String> { let persona = Persona::Tom; - let queue_count = shared.config.scaled(100) as usize; + let queue_count = queue_count(shared.config.scaled(100)); let mut served = Vec::new(); for index in 0..queue_count { served.push(( @@ -277,3 +283,14 @@ pub async fn run(shared: Arc) -> Result<(), String> { producer_pool.close().await; Ok(()) } + +#[cfg(test)] +mod tests { + use super::{queue_count, CRON_JOBS}; + + #[test] + fn every_cron_target_queue_is_served_at_small_scales() { + assert_eq!(queue_count(2), CRON_JOBS); + assert_eq!(queue_count(100), 100); + } +} diff --git a/awa/tests/persona_soak/personas/ui.rs b/awa/tests/persona_soak/personas/ui.rs index d2148fc43..9a47e4e98 100644 --- a/awa/tests/persona_soak/personas/ui.rs +++ b/awa/tests/persona_soak/personas/ui.rs @@ -571,10 +571,7 @@ pub async fn run_ui_operator(shared: Arc) -> Result<(), String> { 5 => { // Bulk actions on the DLQ by filter, as the DLQ page's toolbar does. if rng.chance(50) { - shared.ledger.mark_where( - |record| record.persona == Persona::Mia && record.has(flag::EXHAUSTED), - flag::DLQ_RETRIED, - ); + mia::mark_exhausted_in_queue(&shared, mia::INFER_QUEUE, flag::DLQ_RETRIED); http.post( "/api/dlq/bulk-retry", "/api/dlq/bulk-retry", @@ -582,10 +579,7 @@ pub async fn run_ui_operator(shared: Arc) -> Result<(), String> { ) .await; } else { - shared.ledger.mark_where( - |record| record.persona == Persona::Mia && record.has(flag::EXHAUSTED), - flag::DLQ_PURGED, - ); + mia::mark_exhausted_in_queue(&shared, mia::TRAIN_QUEUE, flag::DLQ_PURGED); http.post( "/api/dlq/bulk-purge", "/api/dlq/bulk-purge", @@ -594,7 +588,8 @@ pub async fn run_ui_operator(shared: Arc) -> Result<(), String> { .await; } } - 6 => { + // Tom declares the cron schedules; without Tom there are none to act on. + 6 if shared.config.has(Persona::Tom) => { let name = format!("soak_tom_cron_{}", rng.below(4)); let paused = http .post( @@ -616,7 +611,7 @@ pub async fn run_ui_operator(shared: Arc) -> Result<(), String> { ) .await; } - 7 => { + 7 if shared.config.has(Persona::Tom) => { let name = format!("soak_tom_cron_{}", rng.below(4)); http.post( "/api/cron/{name}/trigger", diff --git a/awa/tests/persona_soak/python/soak_python.py b/awa/tests/persona_soak/python/soak_python.py index ff6920c3e..b5fe642f8 100644 --- a/awa/tests/persona_soak/python/soak_python.py +++ b/awa/tests/persona_soak/python/soak_python.py @@ -33,6 +33,11 @@ QUEUE_WORK = os.environ.get("SOAK_QUEUE_WORK", "awa_soak_py_work") QUEUE_DEADLINE = os.environ.get("SOAK_QUEUE_DEADLINE", "awa_soak_py_deadline") QUEUE_SHARED = os.environ.get("SOAK_QUEUE_SHARED", "awa_soak_py_shared") +# The harness passes the timings every Rust runtime uses, so a Python maintenance +# leader rescues on the same staleness. +HEARTBEAT_MS = int(os.environ.get("SOAK_HEARTBEAT_MS", "2000")) +STALENESS_MS = int(os.environ.get("SOAK_STALENESS_MS", "12000")) +RESCUE_INTERVAL_MS = os.environ.get("SOAK_RESCUE_INTERVAL_MS") WORK_WORKERS = 8 DEADLINE_WORKERS = 16 SHARED_WORKERS = 6 @@ -325,10 +330,10 @@ async def main(): {"name": QUEUE_DEADLINE, "max_workers": DEADLINE_WORKERS, "deadline_duration_ms": DEADLINE_MS}, {"name": QUEUE_SHARED, "max_workers": SHARED_WORKERS}, ], - heartbeat_interval_ms=2000, - heartbeat_staleness_ms=12000, - heartbeat_rescue_interval_ms=3000, - deadline_rescue_interval_ms=2000, + heartbeat_interval_ms=HEARTBEAT_MS, + heartbeat_staleness_ms=STALENESS_MS, + heartbeat_rescue_interval_ms=int(RESCUE_INTERVAL_MS or 3000), + deadline_rescue_interval_ms=int(RESCUE_INTERVAL_MS or 2000), promote_interval_ms=250, ) # The deadline queue serves a second kind through the same handler. diff --git a/awa/tests/persona_soak/python/test_soak_python.py b/awa/tests/persona_soak/python/test_soak_python.py index 337301719..9736c4c27 100644 --- a/awa/tests/persona_soak/python/test_soak_python.py +++ b/awa/tests/persona_soak/python/test_soak_python.py @@ -1,5 +1,6 @@ """Process-boundary regressions; uses a delayed fake client, never a database.""" +import json import os from pathlib import Path import select @@ -14,6 +15,7 @@ SCRIPT = Path(os.environ.get("SOAK_PYTHON_SCRIPT", Path(__file__).with_name("soak_python.py"))) FAKE_AWA = ''' import asyncio +import json import os from pathlib import Path from types import SimpleNamespace @@ -37,7 +39,8 @@ def task(self, *args, **kwargs): return lambda handler: handler async def start(self, *args, **kwargs): - Path(os.environ["SOAK_TEST_WORKERS"]).touch() + timings = {key: value for key, value in kwargs.items() if key.endswith("_ms")} + Path(os.environ["SOAK_TEST_WORKERS"]).write_text(json.dumps(timings)) async def insert(self, *args, **kwargs): Path(os.environ["SOAK_TEST_PRODUCED"]).touch() @@ -68,8 +71,8 @@ def tearDown(self): self.child.stderr.close() self.directory.cleanup() - def start(self, *, draining=False): - env = os.environ | { + def start(self, *, draining=False, extra_env=None): + env = os.environ | (extra_env or {}) | { "PYTHONPATH": str(self.root), "SOAK_DATABASE_URL": "unused-by-the-fake-client", "SOAK_ROLE": "full", @@ -114,6 +117,22 @@ def test_restart_during_drain_starts_workers_without_producers(self): self.start(draining=True) self.assert_draining_workers() + def test_workers_use_the_timings_the_harness_passes(self): + self.start(extra_env={ + "SOAK_HEARTBEAT_MS": "30000", + "SOAK_STALENESS_MS": "90000", + "SOAK_RESCUE_INTERVAL_MS": "7000", + }) + self.gate.touch() + self.wait_for("READY pid=") + self.assertEqual(json.loads(self.workers.read_text()), { + "heartbeat_interval_ms": 30000, + "heartbeat_staleness_ms": 90000, + "heartbeat_rescue_interval_ms": 7000, + "deadline_rescue_interval_ms": 7000, + "promote_interval_ms": 250, + }) + if __name__ == "__main__": unittest.main() diff --git a/awa/tests/persona_soak/registry.rs b/awa/tests/persona_soak/registry.rs index a0cc42835..2eca256f3 100644 --- a/awa/tests/persona_soak/registry.rs +++ b/awa/tests/persona_soak/registry.rs @@ -78,13 +78,17 @@ impl QueueRegistry { /// Looks a queue up by its physical name; striped queues (`name#N`) resolve to /// their logical queue. pub fn get(&self, name: &str) -> Option> { - let by_name = self.by_name.read().expect("registry lock"); - let index = by_name.get(name).or_else(|| { - let (logical, suffix) = name.rsplit_once('#')?; - suffix.parse::().ok()?; - by_name.get(logical) - })?; - Some(self.queues.read().expect("registry lock")[*index as usize].clone()) + // `register` takes `queues` before `by_name`; holding `by_name` while + // waiting for `queues` would deadlock against it. + let index = { + let by_name = self.by_name.read().expect("registry lock"); + *by_name.get(name).or_else(|| { + let (logical, suffix) = name.rsplit_once('#')?; + suffix.parse::().ok()?; + by_name.get(logical) + })? + }; + Some(self.queues.read().expect("registry lock")[index as usize].clone()) } pub fn by_index(&self, index: u32) -> Option> { @@ -129,3 +133,45 @@ impl Drop for InFlightGuard { } } } + +#[cfg(test)] +mod tests { + use super::QueueRegistry; + use crate::persona_soak::config::Persona; + use std::sync::mpsc; + use std::sync::Arc; + use std::time::Duration; + + #[test] + fn lookups_and_registrations_do_not_deadlock() { + let registry = Arc::new(QueueRegistry::default()); + registry.register("awa_soak_registry_0", Persona::Mia, false); + let (done_tx, done_rx) = mpsc::channel(); + let registering = { + let registry = registry.clone(); + let done_tx = done_tx.clone(); + std::thread::spawn(move || { + for index in 1..20_000 { + registry.register(&format!("awa_soak_registry_{index}"), Persona::Mia, false); + } + let _ = done_tx.send(()); + }) + }; + let looking_up = { + let registry = registry.clone(); + std::thread::spawn(move || { + for _ in 0..200_000 { + assert!(registry.get("awa_soak_registry_0#1").is_some()); + } + let _ = done_tx.send(()); + }) + }; + for _ in 0..2 { + done_rx + .recv_timeout(Duration::from_secs(30)) + .expect("register and get finish without deadlocking"); + } + registering.join().unwrap(); + looking_up.join().unwrap(); + } +} diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index 0725cac04..db00ec8cb 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -8358,7 +8358,7 @@ async fn test_queue_storage_promotion_leaves_a_contended_lane_deferred_and_in_or /// Holds PostgreSQL's cluster-wide notify-queue lock for about `commits` × /// 100 ms. A transaction that issued NOTIFY takes that lock at COMMIT and /// keeps it until its commit record is flushed, which `commit_delay` stretches. -/// Returns once the lock is held. +/// Returns once the holder's own backend holds the lock. async fn hold_notify_commit_lock( pool: &sqlx::PgPool, commits: usize, @@ -8377,6 +8377,10 @@ async fn hold_notify_commit_lock( ) .await .expect("connect the notify holder"); + let holder_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&holder) + .await + .expect("read the notify holder's backend"); let holder_task = tokio::spawn(async move { for _ in 0..commits { sqlx::raw_sql( @@ -8393,8 +8397,9 @@ async fn hold_notify_commit_lock( loop { let held: bool = sqlx::query_scalar( "SELECT EXISTS (SELECT 1 FROM pg_locks WHERE locktype = 'object' \ - AND classid = 'pg_database'::regclass AND objid = 0 AND granted)", + AND classid = 'pg_database'::regclass AND objid = 0 AND granted AND pid = $1)", ) + .bind(holder_pid) .fetch_one(pool) .await .expect("read notify lock"); @@ -8445,7 +8450,7 @@ async fn test_queue_storage_promotion_does_not_wait_for_the_notify_lock_in_its_l enqueue_due(id).await; } - let holder = hold_notify_commit_lock(&pool, 3).await; + let holder = hold_notify_commit_lock(&pool, 10).await; let outcome = store .promote_due_lanes( &pool, @@ -8456,6 +8461,10 @@ async fn test_queue_storage_promotion_does_not_wait_for_the_notify_lock_in_its_l ) .await .expect("promote while the notify lock is held"); + assert!( + !holder.is_finished(), + "the notify holder outlasts the promotion" + ); holder.await.expect("notify holder"); assert_eq!(outcome.promoted, 3, "{outcome:?}"); assert!(outcome.contended.is_empty(), "{outcome:?}"); @@ -8471,7 +8480,7 @@ async fn test_queue_storage_promotion_does_not_wait_for_the_notify_lock_in_its_l .await .expect("connect with a session lock_timeout"); enqueue_due(3).await; - let holder = hold_notify_commit_lock(&pool, 3).await; + let holder = hold_notify_commit_lock(&pool, 10).await; let result = store .promote_due_lanes( &caller_lock_timeout_pool, @@ -8481,6 +8490,10 @@ async fn test_queue_storage_promotion_does_not_wait_for_the_notify_lock_in_its_l &[], ) .await; + assert!( + !holder.is_finished(), + "the notify holder outlasts the promotion" + ); holder.await.expect("notify holder"); caller_lock_timeout_pool.close().await; let outcome = result.expect("a failed wakeup does not fail the promotion"); @@ -17281,6 +17294,55 @@ async fn receipt_admin_ids(pool: &sqlx::PgPool, filter: admin::ListJobsFilter) - .collect() } +/// A per-slot receipt read touches only that slot's claim-ring partitions, so a +/// lock held on another slot's closure evidence, as claim-ring prune takes, does +/// not hold it up. +#[tokio::test] +async fn test_slot_scoped_receipt_reads_do_not_wait_on_other_slots_closures() { + let (_db_guard, pool) = setup_pool(4).await; + let schema = "awa_qs_slot_scoped_closures"; + let store = create_store_with_config(&pool, unique_cancel_store_config(schema)).await; + let queue = "slot_scoped_closures"; + enqueue_job( + &pool, + &store, + &CompleteJob { id: 1 }, + InsertOpts { + queue: queue.into(), + ..Default::default() + }, + ) + .await; + let claimed = store + .claim_runtime_batch(&pool, queue, 1, Duration::ZERO) + .await + .expect("claim into a compact batch"); + assert_eq!(claimed.len(), 1); + let other_slot = (claimed[0].claim.claim_slot + 1) % 2; + + let mut prune = pool.begin().await.expect("begin prune stand-in"); + sqlx::query(&format!( + "LOCK TABLE {schema}.lease_claim_closures_{other_slot}, \ + {schema}.lease_claim_closure_batches_{other_slot} IN ACCESS EXCLUSIVE MODE" + )) + .execute(prune.as_mut()) + .await + .expect("lock the other slot's closure evidence"); + let kinds = tokio::time::timeout( + scaled_timeout(Duration::from_secs(5)), + admin::job_kind_overviews(&pool), + ) + .await + .expect("the read does not wait on another slot's closure locks") + .expect("read kind overviews"); + prune.rollback().await.expect("release the closure locks"); + let complete = kinds + .iter() + .find(|kind| kind.kind == CompleteJob::kind()) + .expect("the running job's kind"); + assert_eq!(complete.job_count, 1); +} + async fn receipt_admin_reads_preserve_scope_and_disposition(receipts: bool, row_local: bool) { let (_db_guard, pool) = setup_pool(6).await; let schema = match (receipts, row_local) { @@ -19512,6 +19574,83 @@ async fn test_historical_tombstoned_claim_uses_emitted_lane_low_to_high() { async fn test_historical_tombstoned_claim_uses_emitted_lane_high_to_low() { moved_tombstoned_claim_lane(4, 2, true, false, false).await; } +/// An attempt whose retained ready bodies are ambiguous stays running. A batch's +/// worth of them, which every pass meets first, must not keep rescue from the stale +/// attempts behind them. +#[tokio::test] +async fn test_ambiguous_attempts_filling_a_batch_do_not_starve_rescue() { + const AMBIGUOUS: i64 = 500; + let (_guard, pool) = setup_pool(4).await; + let schema = "awa_qs_ambiguous_backlog"; + let store = create_store_with_config( + &pool, + QueueStorageConfig { + schema: schema.into(), + lease_claim_receipts: false, + queue_stripe_count: 1, + ..Default::default() + }, + ) + .await; + let queue = "qs_ambiguous_backlog"; + store + .enqueue_batch(&pool, queue, 2, AMBIGUOUS) + .await + .expect("enqueue the ambiguous backlog"); + let claimed = store + .claim_runtime_batch(&pool, queue, AMBIGUOUS, Duration::from_secs(300)) + .await + .expect("claim the ambiguous backlog"); + assert_eq!(claimed.len(), AMBIGUOUS as usize); + // A second body at the same lane position under another priority, as older + // physical aging could leave, leaves no way to choose each attempt's body. + let duplicated = sqlx::query(&format!( + "INSERT INTO {schema}.ready_entries + (ready_slot,ready_generation,job_id,kind,queue,args,priority,attempt,run_lease,max_attempts, + lane_seq,enqueue_shard,run_at,attempted_at,created_at,unique_key,unique_states,payload) + SELECT ready_slot,ready_generation,job_id,kind,queue,args,priority-1,attempt,run_lease, + max_attempts,lane_seq,enqueue_shard,run_at,attempted_at,created_at,unique_key, + unique_states,payload + FROM {schema}.ready_entries WHERE queue=$1" + )) + .bind(queue) + .execute(&pool) + .await + .expect("duplicate the retained bodies"); + assert_eq!(duplicated.rows_affected(), AMBIGUOUS as u64); + + let other_queue = "qs_ambiguous_backlog_other"; + let other_id = enqueue_job( + &pool, + &store, + &CompleteJob { id: 9045 }, + InsertOpts { + queue: other_queue.into(), + ..Default::default() + }, + ) + .await; + let other_claims = store + .claim_runtime_batch(&pool, other_queue, 1, Duration::from_secs(300)) + .await + .expect("claim a recoverable job after the backlog"); + assert_eq!(other_claims.len(), 1); + + let rescued = store + .rescue_stale_heartbeats(&pool, Duration::ZERO) + .await + .expect("ambiguous attempts do not stop rescue of the others"); + assert_eq!( + rescued.iter().map(|job| job.id).collect::>(), + vec![other_id] + ); + assert_eq!( + lease_count(&pool, &store).await, + AMBIGUOUS, + "the ambiguous attempts stay running" + ); +} + #[tokio::test] async fn test_historical_tombstoned_legacy_ambiguous_lane_is_skipped_by_rescue() { moved_tombstoned_claim_lane(2, 4, false, false, false).await; diff --git a/docs/configuration.md b/docs/configuration.md index dae5622ef..c0ec2a062 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -245,7 +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 a dedicated waiter that is served in turn. 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 transaction held a lane lock for longer than 30 s. +- 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 a dedicated waiter that is served in turn. 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 waiter's promotion transaction waited longer than 30 s for one of its locks, usually the lane lock behind the producers queued ahead of it. - 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.