From 78865062ec54d11d23d01019a20f490901ba06ba Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Mon, 9 Nov 2026 00:14:13 +1300 Subject: [PATCH 01/11] test(soak): fix harness races and over-broad oracle exemptions - Budget::reserve replaces a label's own earmark instead of counting it twice. - Dropping a live ClientHost inside the persona runtime shuts its runtime down in the background instead of panicking. - QueueRegistry::get releases the name index before reading the queue list, so it cannot deadlock against register. - Bulk DLQ retry and purge flag only the exhausted jobs on the queue they target. - UiOps skips the cron actions when Tom, which declares the schedules, is not running. - Tom serves at least the queues its cron schedules write to. - Chaos leaves the shard-ops owner, migrator and admin sessions alone. - The unexplained-cancel scan reads every cancellation in its window. - The Python runtime uses the heartbeat, staleness and rescue timings the Rust runtimes use. --- awa/tests/persona_soak/db.rs | 97 ++++++++++++++++--- awa/tests/persona_soak/oracle.rs | 43 +++++++- awa/tests/persona_soak/personas/alex.rs | 12 ++- awa/tests/persona_soak/personas/chaos.rs | 80 ++++++++++++--- awa/tests/persona_soak/personas/mia.rs | 55 +++++++++++ awa/tests/persona_soak/personas/python.rs | 14 +++ awa/tests/persona_soak/personas/tom.rs | 19 +++- awa/tests/persona_soak/personas/ui.rs | 15 +-- awa/tests/persona_soak/python/soak_python.py | 13 ++- .../persona_soak/python/test_soak_python.py | 25 ++++- awa/tests/persona_soak/registry.rs | 60 ++++++++++-- 11 files changed, 379 insertions(+), 54 deletions(-) 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(); + } +} From 8e7487c769c4241defd6028db8f38b8b963952b5 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Mon, 9 Nov 2026 00:14:13 +1300 Subject: [PATCH 02/11] docs(config): describe waiter_timed_out as the waiter's lock wait running out --- docs/configuration.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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. From 5e31f2405d78aa6282ebfbd978adab9cdbb736cc Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Mon, 9 Nov 2026 00:16:29 +1300 Subject: [PATCH 03/11] fix(admin): keep slot-scoped receipt reads off other slots' closure partitions The compact-batch leg of the per-slot open-claim read looked up closure evidence through the parent closure tables, so it locked every slot's closure partitions and waited behind claim-ring prune of another slot. It now reads the selected slot's closure partitions, as the row-local leg already did. --- awa-model/src/queue_storage.rs | 6 +-- awa/tests/queue_storage_runtime_test.rs | 49 +++++++++++++++++++++++++ 2 files changed, 52 insertions(+), 3 deletions(-) diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index 90a774b0b..f977fba22 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 diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index 0725cac04..efe52f25c 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -17281,6 +17281,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) { From ad41f04c5d32e100e4e7c3e4ae0d9d38be3ffa93 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Mon, 9 Nov 2026 01:23:20 +1300 Subject: [PATCH 04/11] fix(queue-storage): let rescue reach attempts behind a batch of ambiguous ones Rescue leaves an attempt whose retained ready bodies are ambiguous running and retries its batch without it, but only within the attempts the pass first selected. Every pass selects the oldest stale attempts first, so a batch's worth of ambiguous ones kept heartbeat, deadline and callback rescue from every attempt behind them. Hydration now reports all of a batch's ambiguous attempts before it writes anything, and the pass sets them aside and selects again. Busy retained attempts still only shrink the pass's selection. --- awa-model/src/queue_storage.rs | 129 +++++++++++++++++------- awa/tests/queue_storage_runtime_test.rs | 77 ++++++++++++++ 2 files changed, 171 insertions(+), 35 deletions(-) diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index f977fba22..76d4c3829 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -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,12 @@ 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) => { + let (job_id, run_lease) = attempts.first().copied().unwrap_or_default(); + AwaError::Validation(format!( + "queue storage ready body ambiguous for deleted lease job {job_id} run_lease {run_lease}" + )) + } HydrateError::Awa(err) => err, } } @@ -2456,10 +2483,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 +10876,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 +10908,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 +11189,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 +11240,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 +11325,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 +15467,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 +15478,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 +15637,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 +15649,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 +15775,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/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index efe52f25c..994cedac9 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -19561,6 +19561,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; From 8fe93d26a91dfc41112af5a60f00b7fd2a41f6fb Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Mon, 9 Nov 2026 01:23:20 +1300 Subject: [PATCH 05/11] test(queue-storage): require the notify holder's own lock to overlap promotion The readiness check accepted any granted notify-queue lock in the cluster, so another notifying backend could let the test run without contention. It now waits for the holder's backend and checks the holder is still committing when promotion returns. --- awa/tests/queue_storage_runtime_test.rs | 21 +++++++++++++++++---- 1 file changed, 17 insertions(+), 4 deletions(-) diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index 994cedac9..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"); From c6a2146a4ca78a2a3b9d7c13581395e5232d3983 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Mon, 9 Nov 2026 01:23:20 +1300 Subject: [PATCH 06/11] fix(worker): clear the attempt's heartbeat and deadline in Completed events The event built from the worker's copy kept the finished attempt's heartbeat_at and deadline_at, which a read of the completed job returns as None. --- awa-worker/src/executor.rs | 2 ++ awa/tests/lifecycle_hook_test.rs | 62 ++++++++++++++++++++++++++++++++ 2 files changed, 64 insertions(+) diff --git a/awa-worker/src/executor.rs b/awa-worker/src/executor.rs index f68955775..458e87962 100644 --- a/awa-worker/src/executor.rs +++ b/awa-worker/src/executor.rs @@ -1839,6 +1839,8 @@ async fn complete_job_queue_storage( updated_job.state = JobState::Completed; updated_job.finalized_at = Some(chrono::Utc::now()); updated_job.progress = None; + updated_job.heartbeat_at = None; + updated_job.deadline_at = None; Ok(CompletionOutcome::Applied { event: Some(UntypedJobEvent::Completed { job: updated_job, diff --git a/awa/tests/lifecycle_hook_test.rs b/awa/tests/lifecycle_hook_test.rs index 3eb72d9ff..68d4bd4ec 100644 --- a/awa/tests/lifecycle_hook_test.rs +++ b/awa/tests/lifecycle_hook_test.rs @@ -183,6 +183,68 @@ 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() { + 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 client = 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(); + } + } + }) + .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)); +} + #[tokio::test] async fn test_typed_started_event_handler_runs() { let _permit = test_gate() From 3dc5b47899d259c83f74ef97f8f9d3eca1d94ed9 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Mon, 9 Nov 2026 01:23:20 +1300 Subject: [PATCH 07/11] docs(changelog): note the rescue, admin kind count and Completed event fixes --- CHANGELOG.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 244d516b3..a8f89e927 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 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 kind counts read each claim slot's receipt closures from that slot's partitions only, so they no longer wait behind claim-ring prune of another slot. +- A `Completed` lifecycle event of a job with a deadline no longer carries 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 From a2872054a5795874a41411194924602f287d2627 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Mon, 9 Nov 2026 01:52:13 +1300 Subject: [PATCH 08/11] fix(worker): give Completed follow-ups the same completed snapshot as the event Completion through Completed follow-up specs built its own snapshot, which kept the finished attempt's heartbeat_at and deadline_at in both the event and the job the specs receive. Both paths now share one helper. --- awa-worker/src/executor.rs | 30 +++++++++---------- awa/tests/lifecycle_hook_test.rs | 49 +++++++++++++++++++++++++++++--- 2 files changed, 60 insertions(+), 19 deletions(-) diff --git a/awa-worker/src/executor.rs b/awa-worker/src/executor.rs index 458e87962..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,17 +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; - updated_job.heartbeat_at = None; - updated_job.deadline_at = 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, @@ -2461,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 68d4bd4ec..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, @@ -187,6 +192,22 @@ async fn test_typed_completed_event_handler_runs() { /// 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 @@ -196,7 +217,8 @@ async fn test_completed_event_of_a_deadline_job_matches_the_terminal_job() { clean_queue(&pool, queue).await; let (tx, mut rx) = mpsc::unbounded_channel(); - let client = Client::builder(pool.clone()) + let follow_up_saw: Arc>> = Arc::default(); + let builder = Client::builder(pool.clone()) .queue( queue, QueueConfig { @@ -213,9 +235,22 @@ async fn test_completed_event_of_a_deadline_job_matches_the_terminal_job() { 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, + } }) - .build() - .unwrap(); + } else { + builder + }; + let client = builder.build().unwrap(); let inserted = awa::insert_with( &pool, @@ -243,6 +278,12 @@ async fn test_completed_event_of_a_deadline_job_matches_the_terminal_job() { (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] From abff44a3f994c3b68a13b1cda4953a30e974c5a1 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Mon, 9 Nov 2026 01:52:13 +1300 Subject: [PATCH 09/11] fix(queue-storage): name every ambiguous attempt count in the hydrate error An empty attempt list printed job 0 run_lease 0, and several attempts named only the first. --- awa-model/src/queue_storage.rs | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index 76d4c3829..6fae9ae27 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -1811,12 +1811,16 @@ impl From for HydrateError { impl From for AwaError { fn from(err: HydrateError) -> Self { match err { - HydrateError::Ambiguous(attempts) => { - let (job_id, run_lease) = attempts.first().copied().unwrap_or_default(); - AwaError::Validation(format!( + 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, } } From 25a2a687ae5ebb38c9ea0fce38d2d83334f0f9b9 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Mon, 9 Nov 2026 01:52:13 +1300 Subject: [PATCH 10/11] docs(changelog): scope the rescue, admin read and Completed snapshot entries --- CHANGELOG.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a8f89e927..c256abad8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,9 +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 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 kind counts read each claim slot's receipt closures from that slot's partitions only, so they no longer wait behind claim-ring prune of another slot. -- A `Completed` lifecycle event of a job with a deadline no longer carries the finished attempt's `heartbeat_at` and `deadline_at`; both are `None`, as a read of the completed job returns them. +- 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 From 9f474d15b2bb595cf2e38c4397ae2c0a39ef3287 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Mon, 9 Nov 2026 01:57:29 +1300 Subject: [PATCH 11/11] ci: give the Postgres service containers 256 MB of shared memory --- .github/workflows/ci.yml | 9 +++++++++ 1 file changed, 9 insertions(+) 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