From 61db389d19eb32ebd41ad6a2acb5b60f6494e14e Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Sat, 7 Nov 2026 15:52:57 +1300 Subject: [PATCH 1/3] Fix claim aging race and retained body hydration --- CHANGELOG.md | 27 +- awa-model/src/queue_storage.rs | 272 +++++++++++-- awa/tests/persona_soak/mod.rs | 1 + awa/tests/persona_soak/personas/abandon.rs | 20 +- awa/tests/persona_soak/run.rs | 2 + awa/tests/persona_soak/worker_errors.rs | 241 +++++++++++ awa/tests/queue_storage_runtime_test.rs | 445 +++++++++++++++++++++ docs/configuration.md | 5 +- 8 files changed, 967 insertions(+), 46 deletions(-) create mode 100644 awa/tests/persona_soak/worker_errors.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index afaaa314f..1732c27f2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,16 +16,22 @@ order; binaries can roll back to 0.6.13 with the patch retained. See the - New queue lanes and their sequences belong to the schema owner and receive the head table's grants, regardless of which runtime creates them. Migration also repairs existing lanes where permitted and reports busy or unauthorized - sequences for a later repair. + sequences for a later repair. `awa storage lane-sequences [--repair]` checks + or repairs them and exits with code 2 while problems remain. - Promotion isolates contended queue/priority lanes so they cannot stall other queues or maintenance. Lock timeouts anywhere in prune or terminal rollup roll back the attempt and report `Blocked`. - Graceful shutdown waits for pending finalization before stopping heartbeat and maintenance, subject to the existing drain timeout. - Reads and callbacks identify priority-aged and moved attempts by job ID and - run lease, excluding tombstoned source rows. Admin lists and kind counts also - include running receipt claims that have not materialized into leases. -- Failed terminal writes preserve the live ready body's arguments, attempt limit and + run lease, preferring live ready bodies over moved source copies. Admin lists + and kind counts also include running receipt claims that have not materialized + into leases. +- Physical priority aging waits for in-flight claims and rechecks claim evidence + before moving a ready row, including claims committed before their cursor + advances. Rescue, cancellation and completion can recover retained tombstoned + bodies left by older aging, preserving claim-time priority metadata. +- Failed terminal writes preserve the ready body's arguments, attempt limit and physical lane. Rust retry and retention carry recover older narrow terminal bodies while their backing remains retained. Stored terminal metadata wins over ready defaults, and counters use the correct stored and replacement @@ -43,9 +49,13 @@ order; binaries can roll back to 0.6.13 with the patch retained. See the ### Added -- Configurable prune lock timeout in Rust and Python, with prune outcome - counters. The timeout applies to each acquisition; queue prune can wait on - up to eight partition locks. The default remains 10 ms. +- `ClientBuilder::prune_lock_timeout` in Rust and + `queue_storage_prune_lock_timeout_ms` in Python configure prune's lock wait. + The timeout applies to each acquisition; queue prune can wait on up to eight + partition locks. The default remains 10 ms. `awa.maintenance.prune.attempts` + includes the outcomes `backed_off` and `error`. +- `awa.maintenance.promote.lane_contention` records promotion lane contention + and waiter outcomes by queue. - Multi-runtime soak coverage for killed workers, shard changes, mixed Rust and Python workloads, durable recovery evidence and ring-lock diagnostics. @@ -65,6 +75,9 @@ order; binaries can roll back to 0.6.13 with the patch retained. See the - The installed `terminal_jobs` SQL view and creation-time bucket joins still use the stored priority for older mismatched rows. Retention carry can widen surviving failed rows, but already-pruned backing bodies cannot be rebuilt. +- Multiple historical tombstoned bodies can remain ambiguous without unique + emitted claim-lane evidence. Hydration reports that ambiguity and rolls back + rather than choosing a body with potentially different arguments or priority. ## [0.6.13] — 2026-11-06 diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index 25bd4a874..b585a30e9 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -1739,6 +1739,18 @@ impl ReadyJobRow { } } +#[derive(Debug, sqlx::FromRow)] +struct ReadyAgingCandidate { + ready_slot: i32, + ready_generation: i64, + queue: String, + priority: i16, + enqueue_shard: i16, + lane_seq: i64, + job_id: i64, + run_lease: i64, +} + #[derive(Debug, Clone, sqlx::FromRow)] struct ReadyTransitionRow { ready_slot: i32, @@ -2357,6 +2369,8 @@ struct ReadySnapshotRow { unique_key: Option>, unique_states: Option, payload: serde_json::Value, + run_lease: i64, + live_body: bool, } #[derive(Debug, Clone, sqlx::FromRow)] @@ -9821,11 +9835,67 @@ impl QueueStorage { let schema = self.schema(); let mut tx = pool.begin().await.map_err(map_sqlx_error)?; + // Claims commit before their cursor advance. Select only a bounded + // set of identities before taking heads, without locking ready rows in + // the reverse order to a claimer. A later statement rechecks these same + // identities after every in-flight claim on their lanes has committed. + let candidates: Vec = sqlx::query_as(&format!( + r#" + SELECT ready.ready_slot, ready.ready_generation, ready.queue, + ready.priority, ready.enqueue_shard, ready.lane_seq, + ready.job_id, ready.run_lease + FROM {schema}.ready_entries AS ready + JOIN {schema}.queue_claim_heads AS claims + ON claims.queue = ready.queue + AND claims.priority = ready.priority + AND claims.enqueue_shard = ready.enqueue_shard + WHERE ready.lane_seq >= {schema}.sequence_next_value(claims.seq_name) + AND ready.priority > 1 + AND ready.run_at <= $1 + AND NOT EXISTS ( + SELECT 1 FROM {schema}.ready_tombstones AS tomb + WHERE tomb.queue = ready.queue + AND tomb.priority = ready.priority + AND tomb.enqueue_shard = ready.enqueue_shard + AND tomb.lane_seq = ready.lane_seq + AND tomb.ready_slot = ready.ready_slot + AND tomb.ready_generation = ready.ready_generation + ) + ORDER BY ready.run_at ASC, ready.lane_seq ASC + LIMIT $2 + "# + )) + .bind(cutoff) + .bind(limit) + .fetch_all(tx.as_mut()) + .await + .map_err(map_sqlx_error)?; + if candidates.is_empty() { + tx.commit().await.map_err(map_sqlx_error)?; + return Ok(Vec::new()); + } + let job_ids: Vec = candidates.iter().map(|row| row.job_id).collect(); + self.lock_claim_heads_of_jobs_tx(&mut tx, &job_ids).await?; + let claimed = self.ready_row_claimed_sql("ready"); + let moved: Vec = sqlx::query_as(&format!( r#" - WITH target AS ( + WITH candidates(ready_slot, ready_generation, queue, priority, + enqueue_shard, lane_seq, job_id, run_lease) AS ( + SELECT * FROM unnest($3::int[], $4::bigint[], $5::text[], $6::smallint[], + $7::smallint[], $8::bigint[], $9::bigint[], $10::bigint[]) + ), target AS ( SELECT ready.* - FROM {schema}.ready_entries AS ready + FROM candidates + JOIN {schema}.ready_entries AS ready + ON ready.ready_slot = candidates.ready_slot + AND ready.ready_generation = candidates.ready_generation + AND ready.queue = candidates.queue + AND ready.priority = candidates.priority + AND ready.enqueue_shard = candidates.enqueue_shard + AND ready.lane_seq = candidates.lane_seq + AND ready.job_id = candidates.job_id + AND ready.run_lease = candidates.run_lease JOIN {schema}.queue_claim_heads AS claims ON claims.queue = ready.queue AND claims.priority = ready.priority @@ -9833,6 +9903,7 @@ impl QueueStorage { WHERE ready.lane_seq >= {schema}.sequence_next_value(claims.seq_name) AND ready.priority > 1 AND ready.run_at <= $1 + AND NOT ({claimed}) AND NOT EXISTS ( SELECT 1 FROM {schema}.ready_tombstones AS tomb WHERE tomb.queue = ready.queue @@ -9878,6 +9949,14 @@ impl QueueStorage { )) .bind(cutoff) .bind(limit) + .bind(candidates.iter().map(|row| row.ready_slot).collect::>()) + .bind(candidates.iter().map(|row| row.ready_generation).collect::>()) + .bind(candidates.iter().map(|row| row.queue.clone()).collect::>()) + .bind(candidates.iter().map(|row| row.priority).collect::>()) + .bind(candidates.iter().map(|row| row.enqueue_shard).collect::>()) + .bind(candidates.iter().map(|row| row.lane_seq).collect::>()) + .bind(&job_ids) + .bind(candidates.iter().map(|row| row.run_lease).collect::>()) .fetch_all(tx.as_mut()) .await .map_err(map_sqlx_error)?; @@ -10620,12 +10699,13 @@ impl QueueStorage { // Resolve moved ready bodies without opening the whole tombstone ring. // Empty input returned above; every remaining slot is validated here. - let tombstone_sources = ready_slots + let slots = ready_slots .iter() .map(|&slot| ring_slot_index(slot, self.queue_slot_count(), "queue")) - .collect::, _>>()? - .into_iter() - .map(|slot| format!("SELECT * FROM {}", ready_tombstone_child_name(schema, slot))) + .collect::, _>>()?; + let tombstone_sources = slots + .iter() + .map(|&slot| format!("SELECT * FROM {}", ready_tombstone_child_name(schema, slot))) .collect::>() .join(" UNION ALL "); @@ -10648,7 +10728,18 @@ impl QueueStorage { ready.created_at, ready.unique_key, ready.unique_states, - COALESCE(ready.payload, '{{}}'::jsonb) AS payload + COALESCE(ready.payload, '{{}}'::jsonb) AS payload, + ready.run_lease, + NOT EXISTS ( + SELECT 1 FROM ({tombstone_sources}) AS tomb + WHERE tomb.ready_slot = ready.ready_slot + AND tomb.ready_generation = ready.ready_generation + AND tomb.queue = ready.queue + AND tomb.priority = ready.priority + AND tomb.enqueue_shard = ready.enqueue_shard + AND tomb.lane_seq = ready.lane_seq + AND tomb.job_id = ready.job_id + ) AS live_body FROM refs JOIN {schema}.ready_entries AS ready ON ready.ready_slot = refs.ready_slot @@ -10658,16 +10749,6 @@ impl QueueStorage { AND ready.lane_seq = refs.lane_seq AND ready.job_id = refs.job_id AND ready.run_lease + 1 = refs.run_lease - WHERE NOT EXISTS ( - SELECT 1 FROM ({tombstone_sources}) AS tomb - WHERE tomb.ready_slot = ready.ready_slot - AND tomb.ready_generation = ready.ready_generation - AND tomb.queue = ready.queue - AND tomb.priority = ready.priority - AND tomb.enqueue_shard = ready.enqueue_shard - AND tomb.lane_seq = ready.lane_seq - AND tomb.job_id = ready.job_id - ) "# )) .bind(&ready_slots) @@ -10681,6 +10762,83 @@ impl QueueStorage { .await .map_err(map_sqlx_error)?; + type ReadyBodyKey = (i32, i64, String, i16, i64, i64, i64); + let mut ready_map: BTreeMap> = BTreeMap::new(); + for row in ready_rows { + ready_map + .entry(( + row.ready_slot, + row.ready_generation, + row.queue.clone(), + row.enqueue_shard, + row.lane_seq, + row.job_id, + row.run_lease + 1, + )) + .or_default() + .push(row); + } + + type ReadyBodyLaneKey = (i32, i64, String, i16, i64, i64, i64, i16); + let mut claimed_lanes: BTreeSet = BTreeSet::new(); + let ambiguous: Vec<&ReadySnapshotRow> = ready_map + .values() + .filter(|rows| rows.len() > 1 && !rows.iter().any(|row| row.live_body)) + .flatten() + .collect(); + if self.lease_claim_receipts() && !ambiguous.is_empty() { + // Normal live bodies and unique retained bodies need no extra + // relation lookup. Only ambiguous tombstoned receipt bodies need + // the emitted physical-lane evidence, scoped to their ready slots. + let attempt_sources = ambiguous + .iter() + .map(|row| row.ready_slot) + .collect::>() + .into_iter() + .map(|slot| ring_slot_index(slot, self.queue_slot_count(), "queue")) + .collect::, _>>()? + .into_iter() + .map(|slot| { + format!( + "SELECT * FROM {}", + ready_claim_attempt_batch_child_name(schema, slot) + ) + }) + .collect::>() + .join(" UNION ALL "); + let lanes: Vec = sqlx::query_as(&format!( + r#" + WITH refs(ready_slot,ready_generation,queue,enqueue_shard,lane_seq,job_id,run_lease,priority) AS ( + SELECT * FROM unnest($1::int[], $2::bigint[], $3::text[], $4::smallint[], + $5::bigint[], $6::bigint[], $7::bigint[], $8::smallint[]) + ) + SELECT ready_slot,ready_generation,queue,enqueue_shard,lane_seq,job_id,run_lease,priority + FROM refs + WHERE EXISTS ( + SELECT 1 FROM ({attempt_sources}) AS attempts + WHERE attempts.ready_slot = refs.ready_slot + AND attempts.ready_generation = refs.ready_generation + AND attempts.queue = refs.queue + AND attempts.priority = refs.priority + AND attempts.enqueue_shard = refs.enqueue_shard + AND attempts.first_lane_seq <= refs.lane_seq + AND attempts.next_lane_seq > refs.lane_seq + AND attempts.lane_ranges @> int8range(refs.lane_seq,refs.lane_seq + 1,'[)') + ) + "# + )) + .bind(ambiguous.iter().map(|row| row.ready_slot).collect::>()) + .bind(ambiguous.iter().map(|row| row.ready_generation).collect::>()) + .bind(ambiguous.iter().map(|row| row.queue.clone()).collect::>()) + .bind(ambiguous.iter().map(|row| row.enqueue_shard).collect::>()) + .bind(ambiguous.iter().map(|row| row.lane_seq).collect::>()) + .bind(ambiguous.iter().map(|row| row.job_id).collect::>()) + .bind(ambiguous.iter().map(|row| row.run_lease + 1).collect::>()) + .bind(ambiguous.iter().map(|row| row.priority).collect::>()) + .fetch_all(tx.as_mut()).await.map_err(map_sqlx_error)?; + claimed_lanes.extend(lanes); + } + let attempt_rows: Vec = sqlx::query_as(&format!( r#" WITH refs(job_id, run_lease) AS ( @@ -10825,23 +10983,6 @@ impl QueueStorage { .await .map_err(map_sqlx_error)?; - let ready_map: BTreeMap<(i32, i64, String, i16, i64, i64), ReadySnapshotRow> = ready_rows - .into_iter() - .map(|row| { - ( - ( - row.ready_slot, - row.ready_generation, - row.queue.clone(), - row.enqueue_shard, - row.lane_seq, - row.job_id, - ), - row, - ) - }) - .collect(); - let attempt_map: BTreeMap<(i64, i64), AttemptStateRow> = attempt_rows .into_iter() .map(|row| ((row.job_id, row.run_lease), row)) @@ -10849,7 +10990,7 @@ impl QueueStorage { let mut hydrated = Vec::with_capacity(deleted.len()); for deleted_row in deleted { - let ready = ready_map + let candidates = ready_map .get(&( deleted_row.ready_slot, deleted_row.ready_generation, @@ -10857,6 +10998,7 @@ impl QueueStorage { deleted_row.enqueue_shard, deleted_row.lane_seq, deleted_row.job_id, + deleted_row.run_lease, )) .ok_or_else(|| { AwaError::Validation(format!( @@ -10864,6 +11006,64 @@ impl QueueStorage { deleted_row.job_id, deleted_row.run_lease )) })?; + // Prefer a live exact-attempt body. A tombstoned body can still + // back an attempt claimed before physical aging moved the source; + // its emitted lane evidence outranks other retained moved bodies. + // Legacy ambiguous bodies have no authoritative physical priority: + // fail instead of silently choosing an unordered source payload. + let rank = |row: &ReadySnapshotRow| { + ( + row.live_body, + claimed_lanes.contains(&( + row.ready_slot, + row.ready_generation, + row.queue.clone(), + row.enqueue_shard, + row.lane_seq, + row.job_id, + row.run_lease + 1, + row.priority, + )), + ) + }; + let strongest = candidates + .iter() + .map(&rank) + .max() + .expect("ready body group is nonempty"); + 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(AwaError::Validation(format!( + "queue storage ready body ambiguous for deleted lease job {} run_lease {}", + deleted_row.job_id, deleted_row.run_lease + ))); + } + // Claim-time aging adds this provenance to the returned JobRow, + // not the immutable ready payload. Recovery must reconstruct the + // same default when it reads that retained body after the claim. + // Mutate the cloned JSON so unrelated payload fields survive. + let mut payload = ready.payload.clone(); + if deleted_row.priority < ready.priority { + let metadata = payload + .as_object_mut() + .ok_or_else(|| { + AwaError::Validation( + "queue storage payload must be a JSON object".to_string(), + ) + })? + .entry("metadata") + .or_insert_with(|| serde_json::json!({})) + .as_object_mut() + .ok_or_else(|| { + AwaError::Validation( + "queue storage payload metadata must be a JSON object".to_string(), + ) + })?; + metadata + .entry("_awa_original_priority") + .or_insert_with(|| serde_json::Value::from(i64::from(ready.priority))); + } let attempt = attempt_map.get(&(deleted_row.job_id, deleted_row.run_lease)); hydrated.push(LeaseTransitionRow { @@ -10886,7 +11086,7 @@ impl QueueStorage { created_at: ready.created_at, unique_key: ready.unique_key.clone(), unique_states: ready.unique_states.clone(), - payload: ready.payload.clone(), + payload, progress: attempt.and_then(|row| row.progress.clone()), }); } diff --git a/awa/tests/persona_soak/mod.rs b/awa/tests/persona_soak/mod.rs index e45c4b7b1..2bccae9a0 100644 --- a/awa/tests/persona_soak/mod.rs +++ b/awa/tests/persona_soak/mod.rs @@ -21,3 +21,4 @@ pub mod shard_oracle; pub mod shared; pub mod stop; pub mod terminal_proof; +pub mod worker_errors; diff --git a/awa/tests/persona_soak/personas/abandon.rs b/awa/tests/persona_soak/personas/abandon.rs index 7cc48f023..6ec671bf4 100644 --- a/awa/tests/persona_soak/personas/abandon.rs +++ b/awa/tests/persona_soak/personas/abandon.rs @@ -124,6 +124,7 @@ pub async fn run_child() { ) else { return; }; + super::super::worker_errors::install(true).expect("child worker error tracing"); let generation: u32 = generation.parse().expect("generation"); let workers: u32 = workers.parse().expect("workers"); let pool_size: u32 = std::env::var(ENV_POOL) @@ -317,10 +318,22 @@ fn spawn_child( .flatten() { let shared = shared.clone(); + let output_guard = super::super::worker_errors::child_output_guard(); tokio::spawn(async move { + let _output_guard = output_guard; let mut lines = BufReader::new(stream).lines(); - while let Ok(Some(line)) = lines.next_line().await { - absorb_output(&shared, &line); + loop { + match lines.next_line().await { + Ok(Some(line)) => absorb_output(&shared, &line), + Ok(None) => break, + Err(error) => { + shared.incomplete( + format!("claiming-child output read failed: {error}"), + vec![], + ); + break; + } + } } }); } @@ -328,6 +341,9 @@ fn spawn_child( } fn absorb_output(shared: &Shared, line: &str) { + if super::super::worker_errors::absorb_child_marker(line) { + return; + } let stats = shared.stats(Persona::ClaimAbandon); let field = |name: &str| { line.split_whitespace() diff --git a/awa/tests/persona_soak/run.rs b/awa/tests/persona_soak/run.rs index a17aaeac3..58992d9e7 100644 --- a/awa/tests/persona_soak/run.rs +++ b/awa/tests/persona_soak/run.rs @@ -17,6 +17,7 @@ use std::time::{Duration, Instant}; const HEADROOM: u32 = 14; pub async fn run_soak(config: SoakConfig) -> Result, String> { + let worker_errors = super::worker_errors::install(false)?; eprintln!( "{}: preparing database {}", report::HEADER, @@ -239,6 +240,7 @@ pub async fn run_soak(config: SoakConfig) -> Result, String> { } server.stop(); } + super::worker_errors::reconcile(&shared, &worker_errors).await; let outcome = Outcome { quiesced, elapsed_seconds: shared.started.elapsed().as_secs_f64(), diff --git a/awa/tests/persona_soak/worker_errors.rs b/awa/tests/persona_soak/worker_errors.rs new file mode 100644 index 000000000..89dfb5070 --- /dev/null +++ b/awa/tests/persona_soak/worker_errors.rs @@ -0,0 +1,241 @@ +//! A structured gate for a known worker validation failure, independent of log text. +//! +//! This subscriber covers Rust runtimes in this process. The separately spawned +//! Rust claiming child forwards only a fixed marker over its existing output pipe. +//! Python-extension internals have no tracing bridge and are not covered by it. + +use super::shared::Shared; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use std::sync::{Arc, OnceLock}; +use std::time::Duration; +use tracing::field::{Field, Visit}; +use tracing::{Event, Subscriber}; +use tracing_subscriber::filter::LevelFilter; +use tracing_subscriber::layer::{Context, SubscriberExt}; +use tracing_subscriber::{Layer, Registry}; + +const MISSING_READY: &str = "ready row missing for deleted lease"; +const AMBIGUOUS_READY: &str = "ready body ambiguous for deleted lease"; +const CHILD_MARKER: &str = "AWA_SOAK_WORKER_ERROR ready_body_unavailable"; +static EVIDENCE: OnceLock, String>> = OnceLock::new(); + +#[derive(Default)] +pub struct WorkerErrors { + invalid_ready_body: AtomicU64, + child_readers: AtomicUsize, +} + +#[derive(Default)] +struct ErrorFields { + invalid_ready_body: bool, +} + +impl Visit for ErrorFields { + fn record_str(&mut self, field: &Field, value: &str) { + if matches!(field.name(), "error" | "message") { + self.invalid_ready_body |= + value.contains(MISSING_READY) || value.contains(AMBIGUOUS_READY); + } + } + + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + if matches!(field.name(), "error" | "message") { + self.invalid_ready_body |= { + let formatted = format!("{value:?}"); + formatted.contains(MISSING_READY) || formatted.contains(AMBIGUOUS_READY) + }; + } + } +} + +#[derive(Clone)] +struct ErrorLayer { + evidence: Arc, + child: bool, +} + +impl Layer for ErrorLayer { + fn on_event(&self, event: &Event<'_>, _context: Context<'_, S>) { + if *event.metadata().level() > tracing::Level::WARN { + return; + } + let mut fields = ErrorFields::default(); + event.record(&mut fields); + if fields.invalid_ready_body { + self.evidence + .invalid_ready_body + .fetch_add(1, Ordering::Relaxed); + if self.child { + // One short synchronous line only for this invariant failure. + // The parent's reader is tracked through EOF before reconciliation. + eprintln!("{CHILD_MARKER}"); + } + } + } +} + +/// Install once, before spawning any worker. An existing unrelated subscriber +/// makes setup fail rather than silently dropping the error gate. +pub fn install(child: bool) -> Result, String> { + EVIDENCE + .get_or_init(|| { + let evidence = Arc::new(WorkerErrors::default()); + let subscriber = Registry::default() + .with( + ErrorLayer { + evidence: evidence.clone(), + child, + } + .with_filter(LevelFilter::WARN), + ) + .with( + tracing_subscriber::fmt::layer() + .with_ansi(false) + .with_writer(std::io::stderr) + .with_filter(LevelFilter::WARN), + ); + tracing::subscriber::set_global_default(subscriber) + .map_err(|error| format!("worker error tracing unavailable: {error}"))?; + Ok(evidence) + }) + .clone() +} + +pub struct ChildOutputGuard(Arc); + +impl Drop for ChildOutputGuard { + fn drop(&mut self) { + self.0.child_readers.fetch_sub(1, Ordering::Release); + } +} + +/// Register before spawning the existing output reader, so a fast child exit +/// cannot race the final pending-reader check. +pub fn child_output_guard() -> ChildOutputGuard { + let evidence = EVIDENCE + .get() + .expect("worker error tracing installed before child spawn") + .as_ref() + .expect("worker error tracing installation succeeded") + .clone(); + evidence.child_readers.fetch_add(1, Ordering::Relaxed); + ChildOutputGuard(evidence) +} + +pub fn absorb_child_marker(line: &str) -> bool { + if line != CHILD_MARKER { + return false; + } + EVIDENCE + .get() + .expect("worker error tracing installed before child spawn") + .as_ref() + .expect("worker error tracing installation succeeded") + .invalid_ready_body + .fetch_add(1, Ordering::Relaxed); + true +} + +/// Call after actor shutdown and the admin child stop, before final report. +pub async fn reconcile(shared: &Shared, evidence: &WorkerErrors) { + let drained = tokio::time::timeout( + crate::ci_timing::scaled_timeout(Duration::from_secs(5)), + async { + while evidence.child_readers.load(Ordering::Acquire) != 0 { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }, + ) + .await + .is_ok(); + if !drained { + shared.incomplete("claiming-child error output did not drain", vec![]); + } + let missing = evidence.invalid_ready_body.load(Ordering::Relaxed); + if missing != 0 { + shared + .run + .add("worker_invalid_ready_body_events", missing as i64); + shared.violation( + "worker_ready_body_invalid", + format!("{missing} worker event(s) reported missing or ambiguous ready bodies; see WARN/ERROR output"), + vec![], + ); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + struct DisplayOnly; + + impl std::fmt::Display for DisplayOnly { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + formatter, + "queue storage {MISSING_READY} job 17 run_lease 2" + ) + } + } + + impl std::fmt::Debug for DisplayOnly { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("debug form omits the error") + } + } + + fn count_events(emit: impl FnOnce()) -> u64 { + let evidence = Arc::new(WorkerErrors::default()); + let subscriber = Registry::default().with(ErrorLayer { + evidence: evidence.clone(), + child: false, + }); + tracing::subscriber::with_default(subscriber, emit); + evidence.invalid_ready_body.load(Ordering::Relaxed) + } + + #[test] + fn display_error_field_is_counted_once_even_when_message_also_matches() { + assert_eq!( + count_events(|| { + tracing::warn!(error = %DisplayOnly, "{MISSING_READY}"); + }), + 1 + ); + } + + #[test] + fn string_error_and_message_only_forms_are_counted() { + assert_eq!( + count_events(|| { + tracing::error!(error = MISSING_READY, "rescue failed"); + tracing::warn!("queue storage {MISSING_READY} job 18 run_lease 3"); + }), + 2 + ); + } + + #[test] + fn ambiguous_ready_body_failure_is_also_gated() { + assert_eq!( + count_events(|| { + tracing::error!(error = AMBIGUOUS_READY, "hydrate failed"); + }), + 1 + ); + } + + #[test] + fn unrelated_chaos_warnings_and_payload_text_are_not_gated() { + assert_eq!( + count_events(|| { + tracing::warn!(error = "canceling statement due to user request", "chaos"); + tracing::error!(error = "lock timeout", "expected contention"); + tracing::warn!(payload = MISSING_READY, "a job argument"); + tracing::info!(error = MISSING_READY, "not a worker failure event"); + }), + 0 + ); + } +} diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index 928a9f13b..3d32f674d 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -18727,3 +18727,448 @@ async fn test_point_read_keeps_located_body_until_failed_carry_can_finish() { assert_eq!(after.errors, before.errors); control.reads.close().await; } + +/// Commit the same installed claim function used by the runtime, deliberately +/// leaving its separate post-commit cursor update unsent. +async fn claim_before_aging_cursor_advance_in_tx( + receipts: bool, + deadline: Duration, +) -> ( + TestDbGuard, + sqlx::PgPool, + QueueStorage, + RawReceiptClaimRow, + sqlx::Transaction<'static, sqlx::Postgres>, +) { + let (guard, pool) = setup_pool(6).await; + let schema = "awa_qs_claimed_aging_body"; + let queue = "qs_claimed_aging_body"; + let store = create_store_with_config( + &pool, + QueueStorageConfig { + schema: schema.into(), + lease_claim_receipts: receipts, + queue_stripe_count: 1, + ..Default::default() + }, + ) + .await; + let job_id = enqueue_job( + &pool, + &store, + &CompleteJob { id: 9041 }, + InsertOpts { + queue: queue.into(), + priority: 4, + max_attempts: 1, + metadata: serde_json::json!({"retained":"claimed-before-aging"}), + ..Default::default() + }, + ) + .await; + backdate_one_lane_ready_job_run_at( + &pool, + schema, + job_id, + Utc::now() - chrono::Duration::minutes(5), + ) + .await; + // Occupy the replacement lane's first position. Old physical aging must + // append this job at seq2, not accidentally leave a live seq1 body that + // matches the already committed claim's original reference. + enqueue_job( + &pool, + &store, + &CompleteJob { id: 9042 }, + InsertOpts { + queue: queue.into(), + priority: 3, + ..Default::default() + }, + ) + .await; + let mut tx = pool.begin().await.expect("begin exact runtime claim stage"); + let claimed: Vec = sqlx::query_as(&format!( + "SELECT ready_slot,ready_generation,job_id,priority,attempt,run_lease,lane_seq,claim_slot + FROM {schema}.claim_ready_runtime($1,1,$2,60::float8)" + )) + .bind(queue) + .bind(deadline.as_secs_f64()) + .fetch_all(tx.as_mut()) + .await + .expect("claim old ready row"); + assert_eq!(claimed.len(), 1); + let original = claimed.into_iter().next().unwrap(); + assert_eq!(original.job_id, job_id); + assert_eq!( + original.priority, 1, + "claim-time effective aging remains enabled" + ); + assert_eq!( + claim_cursor_for(&pool, &store, queue, 4, 0).await, + original.lane_seq + ); + (guard, pool, store, original, tx) +} + +async fn claim_before_aging_cursor_advance( + receipts: bool, + deadline: Duration, +) -> (TestDbGuard, sqlx::PgPool, QueueStorage, RawReceiptClaimRow) { + let (guard, pool, store, claim, tx) = + claim_before_aging_cursor_advance_in_tx(receipts, deadline).await; + tx.commit() + .await + .expect("commit claim before advancing cursor"); + (guard, pool, store, claim) +} + +async fn assert_claimed_aging_rescue_body( + pool: &sqlx::PgPool, + store: &QueueStorage, + claim: &RawReceiptClaimRow, +) { + let rescued = store + .rescue_stale_heartbeats(pool, Duration::ZERO) + .await + .expect("hydrate retained claimed body"); + assert_eq!(rescued.len(), 1); + assert_eq!(rescued[0].id, claim.job_id); + assert_eq!(rescued[0].state, JobState::Failed); + assert_eq!(rescued[0].args, serde_json::json!({"id":9041})); + assert_eq!(rescued[0].max_attempts, 1); + assert_eq!( + rescued[0].priority, 4, + "terminal body keeps its physical lane" + ); + assert_eq!(rescued[0].metadata["retained"], "claimed-before-aging"); + assert_eq!(rescued[0].metadata["_awa_original_priority"], 4); + let retried = store + .retry_job(pool, claim.job_id) + .await + .expect("retry rescued body") + .expect("failed attempt retries"); + assert_eq!(retried.args, rescued[0].args); + assert_eq!(retried.max_attempts, 1); + assert_eq!(retried.priority, 4); + assert_eq!(retried.metadata["_awa_original_priority"], 4); +} + +async fn aging_skips_committed_claim_before_cursor(receipts: bool, deadline: Duration) { + let (_guard, pool, store, claim) = claim_before_aging_cursor_advance(receipts, deadline).await; + let aged = store + .age_waiting_priorities(&pool, Duration::from_secs(60), 1) + .await + .expect("age with lagging claim cursor"); + assert!( + aged.is_empty(), + "a committed claim is not waiting, even before cursor advancement: {aged:?}" + ); + assert_eq!(ready_tombstone_count(&pool, &store).await, 0); + assert_claimed_aging_rescue_body(&pool, &store, &claim).await; +} + +async fn historical_tombstoned_claim_hydrates(receipts: bool, deadline: Duration) { + let (_guard, pool, store, claim) = claim_before_aging_cursor_advance(receipts, deadline).await; + // Reproduce the retained source state written by pre-fix physical aging. + // Its replacement ready row has a different lane_seq and therefore cannot + // back this exact claim. Keep that unrelated body rather than deleting it. + let schema = store.schema(); + let mut tx = pool.begin().await.expect("begin historical aging shape"); + sqlx::query(&format!( + "INSERT INTO {schema}.ready_tombstones + (ready_slot,ready_generation,queue,priority,enqueue_shard,lane_seq,job_id) + SELECT ready_slot,ready_generation,queue,priority,enqueue_shard,lane_seq,job_id + FROM {schema}.ready_entries WHERE job_id=$1" + )) + .bind(claim.job_id) + .execute(tx.as_mut()) + .await + .expect("historical source tombstone"); + 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, + {schema}.reserve_enqueue_seq(queue,(priority-1)::smallint,enqueue_shard,1::bigint),enqueue_shard, + run_at,attempted_at,created_at,unique_key,unique_states, + jsonb_set(COALESCE(payload,'{{}}'::jsonb),'{{metadata}}', + COALESCE(payload->'metadata','{{}}'::jsonb) + || jsonb_build_object('_awa_original_priority',priority)) + FROM {schema}.ready_entries WHERE job_id=$1" + )).bind(claim.job_id).execute(tx.as_mut()).await.expect("historical replacement at different position"); + tx.commit() + .await + .expect("commit historical source and replacement"); + let positions: Vec = sqlx::query_scalar(&format!( + "SELECT lane_seq FROM {schema}.ready_entries WHERE job_id=$1 ORDER BY priority DESC" + )) + .bind(claim.job_id) + .fetch_all(&pool) + .await + .expect("source/replacement positions"); + assert_eq!(positions, vec![1, 2]); + let provenance: Vec<(i16, Option)> = sqlx::query_as(&format!( + "SELECT priority,(payload->'metadata'->>'_awa_original_priority')::bigint + FROM {schema}.ready_entries WHERE job_id=$1 ORDER BY priority DESC" + )) + .bind(claim.job_id) + .fetch_all(&pool) + .await + .expect("old writer provenance placement"); + assert_eq!( + provenance, + vec![(4, None), (3, Some(4))], + "old physical aging adds provenance only to its replacement body" + ); + assert_claimed_aging_rescue_body(&pool, &store, &claim).await; +} + +#[tokio::test] +async fn test_physical_aging_skips_committed_compact_claim_before_cursor() { + aging_skips_committed_claim_before_cursor(true, Duration::ZERO).await; +} +#[tokio::test] +async fn test_physical_aging_skips_committed_deadline_claim_before_cursor() { + aging_skips_committed_claim_before_cursor(true, Duration::from_secs(300)).await; +} +#[tokio::test] +async fn test_physical_aging_skips_committed_legacy_claim_before_cursor() { + aging_skips_committed_claim_before_cursor(false, Duration::from_secs(300)).await; +} +#[tokio::test] +async fn test_historical_aged_tombstoned_compact_claim_rescues_body() { + historical_tombstoned_claim_hydrates(true, Duration::ZERO).await; +} +#[tokio::test] +async fn test_historical_aged_tombstoned_deadline_claim_rescues_body() { + historical_tombstoned_claim_hydrates(true, Duration::from_secs(300)).await; +} +#[tokio::test] +async fn test_historical_aged_tombstoned_legacy_claim_rescues_body() { + historical_tombstoned_claim_hydrates(false, Duration::from_secs(300)).await; +} +async fn moved_tombstoned_claim_lane(from: i16, to: i16, receipts: bool) { + let (_guard, pool) = setup_pool(6).await; + let schema = "awa_qs_moved_tomb_claim"; + let queue = "qs_moved_tomb_claim"; + let store = create_store_with_config( + &pool, + QueueStorageConfig { + schema: schema.into(), + lease_claim_receipts: receipts, + queue_stripe_count: 1, + ..Default::default() + }, + ) + .await; + let id = enqueue_job(&pool, &store, &CompleteJob { id: 9043 }, InsertOpts { + queue: queue.into(), priority: from, max_attempts: 1, + metadata: serde_json::json!({"retained":"moved-tomb-claim", "_awa_original_priority":from}), + ..Default::default() + }).await; + backdate_one_lane_ready_job_run_at( + &pool, + schema, + id, + Utc::now() - chrono::Duration::minutes(5), + ) + .await; + assert!(store + .set_priority(&pool, id, to) + .await + .expect("move before claim")); + let body_positions: Vec<(i16, i64)> = sqlx::query_as(&format!( + "SELECT priority,lane_seq FROM {schema}.ready_entries WHERE job_id=$1 ORDER BY priority" + )) + .bind(id) + .fetch_all(&pool) + .await + .expect("shared moved positions"); + assert_eq!(body_positions.len(), 2); + assert!(body_positions.iter().all(|row| row.1 == 1)); + let mut claims = Vec::new(); + for _ in 0..3 { + claims = sqlx::query_as::<_, RawReceiptClaimRow>(&format!( + "SELECT ready_slot,ready_generation,job_id,priority,attempt,run_lease,lane_seq,claim_slot + FROM {schema}.claim_ready_runtime($1,1,300::float8,60::float8)" + )).bind(queue).fetch_all(&pool).await.expect("claim moved body before cursor update"); + if !claims.is_empty() { + break; + } + // A higher-priority tombstoned source may first consume its spent + // cursor prefix; the actual target claim still omits cursor advance. + } + assert_eq!(claims.len(), 1); + assert_eq!(claims[0].job_id, id); + assert_eq!( + claims[0].priority, 1, + "effective priority cannot identify either physical lane" + ); + // The old source was tombstoned by the real move. Model old physical aging + // tombstoning the claimed source as well; both exact-ref bodies now remain. + tombstone_ready_job(&pool, &store, id).await; + assert_eq!(ready_tombstone_count(&pool, &store).await, 2); + if !receipts { + let err = store + .rescue_stale_heartbeats(&pool, Duration::ZERO) + .await + .expect_err("legacy ambiguous physical lanes must not be selected arbitrarily"); + assert!(err.to_string().contains("ready body ambiguous"), "{err}"); + assert_eq!( + lease_count(&pool, &store).await, + 1, + "ambiguous mutation rolls back" + ); + return; + } + let emitted: Vec = sqlx::query_scalar(&format!( + "SELECT priority FROM {schema}.ready_claim_attempt_batches WHERE queue=$1" + )) + .bind(queue) + .fetch_all(&pool) + .await + .expect("physical emitted lane proof"); + assert_eq!(emitted, vec![to]); + let rescued = store + .rescue_stale_heartbeats(&pool, Duration::ZERO) + .await + .expect("use emitted physical lane for retained tombstoned body"); + assert_eq!(rescued.len(), 1); + assert_eq!(rescued[0].priority, to); + assert_eq!(rescued[0].args, serde_json::json!({"id":9043})); + assert_eq!(rescued[0].max_attempts, 1); + assert_eq!(rescued[0].metadata["_awa_original_priority"], from); + let retried = store + .retry_job(&pool, id) + .await + .expect("retry selected moved body") + .unwrap(); + assert_eq!(retried.priority, to); + assert_eq!(retried.args, rescued[0].args); + assert_eq!(retried.max_attempts, 1); + assert_eq!(retried.metadata["_awa_original_priority"], from); +} + +#[tokio::test] +async fn test_historical_tombstoned_claim_uses_emitted_lane_low_to_high() { + moved_tombstoned_claim_lane(2, 4, true).await; +} +#[tokio::test] +async fn test_historical_tombstoned_claim_uses_emitted_lane_high_to_low() { + moved_tombstoned_claim_lane(4, 2, true).await; +} +#[tokio::test] +async fn test_historical_tombstoned_legacy_ambiguous_lanes_roll_back() { + moved_tombstoned_claim_lane(2, 4, false).await; +} + +#[tokio::test] +async fn test_historical_tombstoned_legacy_claim_completes_body() { + let (_guard, pool, store, claim) = + claim_before_aging_cursor_advance(false, Duration::from_secs(300)).await; + tombstone_ready_job(&pool, &store, claim.job_id).await; + let updated = store + .complete_job_batch_by_id(&pool, &[(claim.job_id, claim.run_lease)]) + .await + .expect("complete an already tombstoned claimed body"); + assert_eq!(updated, vec![(claim.job_id, claim.run_lease)]); + let done = store + .load_job(&pool, claim.job_id) + .await + .expect("read completion") + .unwrap(); + assert_eq!(done.state, JobState::Completed); + assert_eq!(done.args, serde_json::json!({"id":9041})); + assert_eq!(done.max_attempts, 1); + assert_eq!(done.priority, 4); + assert_eq!(done.metadata["retained"], "claimed-before-aging"); + assert_eq!(done.metadata["_awa_original_priority"], 4); +} + +async fn tombstoned_claim_cancels_body(receipts: bool, deadline: Duration) { + let (_guard, pool, store, claim) = claim_before_aging_cursor_advance(receipts, deadline).await; + tombstone_ready_job(&pool, &store, claim.job_id).await; + let done = store + .cancel_running( + &pool, + claim.job_id, + claim.run_lease, + "cancel retained body", + None, + ) + .await + .expect("cancel an already tombstoned claimed body") + .unwrap(); + assert_eq!(done.state, JobState::Cancelled); + assert_eq!(done.args, serde_json::json!({"id":9041})); + assert_eq!(done.max_attempts, 1); + assert_eq!(done.priority, 4); + assert_eq!(done.metadata["retained"], "claimed-before-aging"); + assert_eq!(done.metadata["_awa_original_priority"], 4); +} +#[tokio::test] +async fn test_historical_tombstoned_compact_claim_cancels_body() { + tombstoned_claim_cancels_body(true, Duration::ZERO).await; +} +#[tokio::test] +async fn test_historical_tombstoned_deadline_claim_cancels_body() { + tombstoned_claim_cancels_body(true, Duration::from_secs(300)).await; +} +#[tokio::test] +async fn test_historical_tombstoned_legacy_claim_cancels_body() { + tombstoned_claim_cancels_body(false, Duration::from_secs(300)).await; +} + +#[tokio::test] +async fn test_physical_aging_waits_for_claim_head_then_uses_fresh_snapshot() { + let (_guard, pool, store, claim, mut claim_tx) = + claim_before_aging_cursor_advance_in_tx(true, Duration::ZERO).await; + let holder: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(claim_tx.as_mut()) + .await + .expect("actual claim holder PID"); + let aging_pool = PgPoolOptions::new() + .max_connections(1) + .connect_with( + (*pool.connect_options()) + .clone() + .application_name("awa_aging_head_wait_control"), + ) + .await + .expect("separate aging connection"); + let age = store.age_waiting_priorities(&aging_pool, Duration::from_secs(60), 1); + let release = async { + tokio::time::timeout(ci_timing::scaled_timeout(Duration::from_secs(5)), async { + loop { + let waiting: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM pg_stat_activity + WHERE application_name='awa_aging_head_wait_control' + AND wait_event_type='Lock' AND $1=ANY(pg_blocking_pids(pid)))", + ) + .bind(holder) + .fetch_one(&pool) + .await + .expect("observe actual claim-head wait"); + if waiting { + break; + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .expect("aging must wait for the actual uncommitted claim head"); + claim_tx + .commit() + .await + .expect("publish claim while aging waits"); + }; + let (aged, ()) = tokio::join!(age, release); + assert!( + aged.expect("age after committed claim").is_empty(), + "the statement after the head lock must see the committed claim" + ); + assert_eq!(ready_tombstone_count(&pool, &store).await, 0); + assert_claimed_aging_rescue_body(&pool, &store, &claim).await; + aging_pool.close().await; +} diff --git a/docs/configuration.md b/docs/configuration.md index a93285501..d461e3db3 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -582,7 +582,10 @@ Use the defaults unless you have a reason not to: - Increase `claim_slot_count` if the rotation cadence (`claim_rotate_interval`) plus the slot count combine to a partition retention window shorter than your longest in-flight zero-deadline short job; running out of empty slots forces `rotate_claims` to return `SkippedBusy` and the receipt-plane churn falls back onto a smaller working set of partitions. - Increase `queue_stripe_count` only for measured workloads where many small replicas contend on the same logical queue. This is queue-storage configuration, not a per-queue `QueueConfig` field: it gives each logical queue the same number of physical stripes. Striping spreads a hot logical queue over `queue#N` physical coordination paths, but it weakens perfect global ordering and can regress calmer shapes if overused. For single hot-queue workloads, benchmark `2` stripes first; higher values should be treated as workload-specific tuning rather than a general default. - Increase `prune_lock_timeout` when `increase(awa_maintenance_prune_attempts_total{awa_ring_outcome="blocked"}[15m])` stays above zero and the rings stop recycling, typically because frequent short reads of the ring partitions (admin or dashboard queries over the queue-storage tables) hold a conflicting lock at the instant prune tries. A stalled ring grows its current slot, which makes those queries slower and the contention worse. Raise it in small steps (for example `50ms`, then `100ms`, then `250ms`) and keep it well below PostgreSQL's `deadlock_timeout` (default `1s`) and below any `statement_timeout` on the pool's connections. A single lock wait that reaches `deadlock_timeout` makes PostgreSQL run its deadlock detector, and if claims or enqueues are part of a lock cycle with it, the detector can cancel them rather than the prune; a short `lock_timeout` abandons the prune first, so the prune is always the one that gives way. A `statement_timeout` at or below the lock timeout cancels the prune statement before the lock timeout fires, so it fails with an error instead of reporting `blocked`. Startup logs a warning when the configured timeout is at or above either setting. While prune waits, new accesses to the same partition also queue behind its lock request, so a longer timeout trades claim latency for reclamation. Shortening the readers is the more durable fix when they run longer than a few hundred milliseconds. Watch `awa_maintenance_prune_attempts_total` by `awa_ring` and `awa_ring_outcome`: `pruned` should advance steadily, `blocked` means the lock timeout expired, `skipped_active` (with `awa_ring_reason`) means live rows kept the slot, `backed_off` means prune was skipped for up to 32 ticks after consecutive `skipped_active` or `blocked` results, so a persistently blocked ring is retried less often over time, and `error` is a database failure. -- Increase rotation intervals to reduce partition churn and metadata activity. +- Consider increasing rotation intervals before raising `prune_lock_timeout`: + longer intervals can reduce how often prune requests exclusive locks, while + a longer timeout lets each request wait longer. Confirm the effect with + workload measurements. - Decrease rotation intervals to tighten dead-tuple bounds at the cost of more frequent rotate/prune work. Longer rotation intervals accumulate more work in each slot. At a steady enqueue From f473016c95a3d0ba0820d17c1e7b653ce3f27913 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Sat, 7 Nov 2026 16:18:17 +1300 Subject: [PATCH 2/3] Preserve claimed bodies across historical aging collisions --- CHANGELOG.md | 12 ++- awa-model/src/queue_storage.rs | 61 ++++++----- awa/tests/queue_storage_runtime_test.rs | 132 ++++++++++++++++++------ 3 files changed, 144 insertions(+), 61 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1732c27f2..001f7b41a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,7 +30,9 @@ order; binaries can roll back to 0.6.13 with the patch retained. See the - Physical priority aging waits for in-flight claims and rechecks claim evidence before moving a ready row, including claims committed before their cursor advances. Rescue, cancellation and completion can recover retained tombstoned - bodies left by older aging, preserving claim-time priority metadata. + bodies left by older aging, preserving claim-time priority metadata. Emitted + claim-lane evidence takes precedence over colliding live replacements; those + recovered terminal bodies remain self-contained through retry. - Failed terminal writes preserve the ready body's arguments, attempt limit and physical lane. Rust retry and retention carry recover older narrow terminal bodies while their backing remains retained. Stored terminal metadata wins @@ -75,9 +77,11 @@ order; binaries can roll back to 0.6.13 with the patch retained. See the - The installed `terminal_jobs` SQL view and creation-time bucket joins still use the stored priority for older mismatched rows. Retention carry can widen surviving failed rows, but already-pruned backing bodies cannot be rebuilt. -- Multiple historical tombstoned bodies can remain ambiguous without unique - emitted claim-lane evidence. Hydration reports that ambiguity and rolls back - rather than choosing a body with potentially different arguments or priority. +- Legacy or backfilled claims can lack unique emitted claim-lane evidence. + Recovery then prefers a unique live body, which cannot reliably identify the + original physical lane if an old aging race left a colliding live replacement. + Multiple equally supported bodies cause an explicit ambiguity error and + transaction rollback. ## [0.6.13] — 2026-11-06 diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index b585a30e9..3a7ba35ad 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -2408,6 +2408,7 @@ struct LeaseTransitionRow { unique_states: Option, payload: serde_json::Value, progress: Option, + self_contained_terminal: bool, } impl LeaseTransitionRow { @@ -2431,7 +2432,13 @@ impl LeaseTransitionRow { attempt: self.attempt, run_lease: self.run_lease, max_attempts: self.max_attempts, - lane_seq: self.lane_seq, + // A competing live copy must not replace this recovered body + // when later terminal reads resolve retained ready rows. + lane_seq: if self.self_contained_terminal { + -self.job_id + } else { + self.lane_seq + }, enqueue_shard: self.enqueue_shard, run_at: self.run_at, attempted_at: self.attempted_at, @@ -10783,13 +10790,13 @@ impl QueueStorage { let mut claimed_lanes: BTreeSet = BTreeSet::new(); let ambiguous: Vec<&ReadySnapshotRow> = ready_map .values() - .filter(|rows| rows.len() > 1 && !rows.iter().any(|row| row.live_body)) + .filter(|rows| rows.len() > 1) .flatten() .collect(); - if self.lease_claim_receipts() && !ambiguous.is_empty() { - // Normal live bodies and unique retained bodies need no extra - // relation lookup. Only ambiguous tombstoned receipt bodies need - // the emitted physical-lane evidence, scoped to their ready slots. + if !ambiguous.is_empty() { + // A unique body needs no extra relation lookup. Multiple bodies + // need the emitted physical lane even if a replacement is live; + // scope that evidence to the referenced ready slots. let attempt_sources = ambiguous .iter() .map(|row| row.ready_slot) @@ -11006,25 +11013,28 @@ impl QueueStorage { deleted_row.job_id, deleted_row.run_lease )) })?; - // Prefer a live exact-attempt body. A tombstoned body can still - // back an attempt claimed before physical aging moved the source; - // its emitted lane evidence outranks other retained moved bodies. - // Legacy ambiguous bodies have no authoritative physical priority: - // fail instead of silently choosing an unordered source payload. + // Emitted physical-lane evidence wins over a live replacement. + // Without it, retain the live-body compatibility fallback; it + // cannot identify a legacy claim's original colliding lane. + let emitted = |row: &ReadySnapshotRow| { + claimed_lanes.contains(&( + row.ready_slot, + row.ready_generation, + row.queue.clone(), + row.enqueue_shard, + row.lane_seq, + row.job_id, + row.run_lease + 1, + row.priority, + )) + }; + let has_emitted = candidates.iter().any(&emitted); let rank = |row: &ReadySnapshotRow| { - ( - row.live_body, - claimed_lanes.contains(&( - row.ready_slot, - row.ready_generation, - row.queue.clone(), - row.enqueue_shard, - row.lane_seq, - row.job_id, - row.run_lease + 1, - row.priority, - )), - ) + if has_emitted { + emitted(row) + } else { + row.live_body + } }; let strongest = candidates .iter() @@ -11088,6 +11098,9 @@ impl QueueStorage { unique_states: ready.unique_states.clone(), payload, progress: attempt.and_then(|row| row.progress.clone()), + self_contained_terminal: has_emitted + && !ready.live_body + && candidates.iter().any(|row| row.live_body), }); } diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index 3d32f674d..4ac81b712 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -18733,6 +18733,7 @@ async fn test_point_read_keeps_located_body_until_failed_carry_can_finish() { async fn claim_before_aging_cursor_advance_in_tx( receipts: bool, deadline: Duration, + occupy_replacement_lane: bool, ) -> ( TestDbGuard, sqlx::PgPool, @@ -18773,20 +18774,28 @@ async fn claim_before_aging_cursor_advance_in_tx( Utc::now() - chrono::Duration::minutes(5), ) .await; - // Occupy the replacement lane's first position. Old physical aging must - // append this job at seq2, not accidentally leave a live seq1 body that - // matches the already committed claim's original reference. - enqueue_job( - &pool, - &store, - &CompleteJob { id: 9042 }, - InsertOpts { - queue: queue.into(), - priority: 3, - ..Default::default() - }, - ) - .await; + if occupy_replacement_lane { + // Occupy the replacement lane's first position. Old physical aging must + // append this job at seq2, not accidentally leave a live seq1 body that + // matches the already committed claim's original reference. + enqueue_job( + &pool, + &store, + &CompleteJob { id: 9042 }, + InsertOpts { + queue: queue.into(), + priority: 3, + ..Default::default() + }, + ) + .await; + } + if !occupy_replacement_lane { + store + .prepare_queue(&pool, queue, 1) + .await + .expect("prepare empty replacement lane"); + } let mut tx = pool.begin().await.expect("begin exact runtime claim stage"); let claimed: Vec = sqlx::query_as(&format!( "SELECT ready_slot,ready_generation,job_id,priority,attempt,run_lease,lane_seq,claim_slot @@ -18816,7 +18825,7 @@ async fn claim_before_aging_cursor_advance( deadline: Duration, ) -> (TestDbGuard, sqlx::PgPool, QueueStorage, RawReceiptClaimRow) { let (guard, pool, store, claim, tx) = - claim_before_aging_cursor_advance_in_tx(receipts, deadline).await; + claim_before_aging_cursor_advance_in_tx(receipts, deadline, true).await; tx.commit() .await .expect("commit claim before advancing cursor"); @@ -18827,6 +18836,7 @@ async fn assert_claimed_aging_rescue_body( pool: &sqlx::PgPool, store: &QueueStorage, claim: &RawReceiptClaimRow, + self_contained_terminal: bool, ) { let rescued = store .rescue_stale_heartbeats(pool, Duration::ZERO) @@ -18843,6 +18853,26 @@ async fn assert_claimed_aging_rescue_body( ); assert_eq!(rescued[0].metadata["retained"], "claimed-before-aging"); assert_eq!(rescued[0].metadata["_awa_original_priority"], 4); + let schema = store.schema(); + let terminal: (i64, bool) = sqlx::query_as(&format!( + "SELECT lane_seq,args IS NOT NULL FROM {schema}.done_entries WHERE job_id=$1" + )) + .bind(claim.job_id) + .fetch_one(pool) + .await + .expect("terminal backing identity"); + assert_eq!( + terminal, + ( + if self_contained_terminal { + -claim.job_id + } else { + claim.lane_seq + }, + self_contained_terminal, + ), + "only conflicting historical body recovery needs a self-contained terminal row" + ); let retried = store .retry_job(pool, claim.job_id) .await @@ -18865,14 +18895,20 @@ async fn aging_skips_committed_claim_before_cursor(receipts: bool, deadline: Dur "a committed claim is not waiting, even before cursor advancement: {aged:?}" ); assert_eq!(ready_tombstone_count(&pool, &store).await, 0); - assert_claimed_aging_rescue_body(&pool, &store, &claim).await; + assert_claimed_aging_rescue_body(&pool, &store, &claim, false).await; } -async fn historical_tombstoned_claim_hydrates(receipts: bool, deadline: Duration) { - let (_guard, pool, store, claim) = claim_before_aging_cursor_advance(receipts, deadline).await; - // Reproduce the retained source state written by pre-fix physical aging. - // Its replacement ready row has a different lane_seq and therefore cannot - // back this exact claim. Keep that unrelated body rather than deleting it. +async fn historical_tombstoned_claim_hydrates( + receipts: bool, + deadline: Duration, + colliding_position: bool, +) { + let (_guard, pool, store, claim, tx) = + claim_before_aging_cursor_advance_in_tx(receipts, deadline, !colliding_position).await; + tx.commit() + .await + .expect("commit claim before old physical aging"); + // Reproduce the retained source and live replacement written by old aging. let schema = store.schema(); let mut tx = pool.begin().await.expect("begin historical aging shape"); sqlx::query(&format!( @@ -18896,7 +18932,7 @@ async fn historical_tombstoned_claim_hydrates(receipts: bool, deadline: Duration COALESCE(payload->'metadata','{{}}'::jsonb) || jsonb_build_object('_awa_original_priority',priority)) FROM {schema}.ready_entries WHERE job_id=$1" - )).bind(claim.job_id).execute(tx.as_mut()).await.expect("historical replacement at different position"); + )).bind(claim.job_id).execute(tx.as_mut()).await.expect("historical replacement body"); tx.commit() .await .expect("commit historical source and replacement"); @@ -18907,7 +18943,7 @@ async fn historical_tombstoned_claim_hydrates(receipts: bool, deadline: Duration .fetch_all(&pool) .await .expect("source/replacement positions"); - assert_eq!(positions, vec![1, 2]); + assert_eq!(positions, vec![1, if colliding_position { 1 } else { 2 }]); let provenance: Vec<(i16, Option)> = sqlx::query_as(&format!( "SELECT priority,(payload->'metadata'->>'_awa_original_priority')::bigint FROM {schema}.ready_entries WHERE job_id=$1 ORDER BY priority DESC" @@ -18921,7 +18957,7 @@ async fn historical_tombstoned_claim_hydrates(receipts: bool, deadline: Duration vec![(4, None), (3, Some(4))], "old physical aging adds provenance only to its replacement body" ); - assert_claimed_aging_rescue_body(&pool, &store, &claim).await; + assert_claimed_aging_rescue_body(&pool, &store, &claim, colliding_position).await; } #[tokio::test] @@ -18938,17 +18974,26 @@ async fn test_physical_aging_skips_committed_legacy_claim_before_cursor() { } #[tokio::test] async fn test_historical_aged_tombstoned_compact_claim_rescues_body() { - historical_tombstoned_claim_hydrates(true, Duration::ZERO).await; + historical_tombstoned_claim_hydrates(true, Duration::ZERO, false).await; } #[tokio::test] async fn test_historical_aged_tombstoned_deadline_claim_rescues_body() { - historical_tombstoned_claim_hydrates(true, Duration::from_secs(300)).await; + historical_tombstoned_claim_hydrates(true, Duration::from_secs(300), false).await; } #[tokio::test] async fn test_historical_aged_tombstoned_legacy_claim_rescues_body() { - historical_tombstoned_claim_hydrates(false, Duration::from_secs(300)).await; + historical_tombstoned_claim_hydrates(false, Duration::from_secs(300), false).await; +} +#[tokio::test] +async fn test_historical_colliding_live_replacement_compact_rescues_claimed_body() { + historical_tombstoned_claim_hydrates(true, Duration::ZERO, true).await; +} +#[tokio::test] +async fn test_historical_colliding_live_replacement_deadline_rescues_claimed_body() { + historical_tombstoned_claim_hydrates(true, Duration::from_secs(300), true).await; } -async fn moved_tombstoned_claim_lane(from: i16, to: i16, receipts: bool) { + +async fn moved_tombstoned_claim_lane(from: i16, to: i16, receipts: bool, reopen_as_legacy: bool) { let (_guard, pool) = setup_pool(6).await; let schema = "awa_qs_moved_tomb_claim"; let queue = "qs_moved_tomb_claim"; @@ -19005,6 +19050,22 @@ async fn moved_tombstoned_claim_lane(from: i16, to: i16, receipts: bool) { claims[0].priority, 1, "effective priority cannot identify either physical lane" ); + let store = if reopen_as_legacy { + store + .register_callback(&pool, id, claims[0].run_lease, Duration::from_secs(60)) + .await + .expect("materialize receipt before reopening"); + assert_eq!(lease_count(&pool, &store).await, 1); + QueueStorage::new(QueueStorageConfig { + schema: schema.into(), + lease_claim_receipts: false, + queue_stripe_count: 1, + ..Default::default() + }) + .expect("reopen existing custom store in legacy mode") + } else { + store + }; // The old source was tombstoned by the real move. Model old physical aging // tombstoning the claimed source as well; both exact-ref bodies now remain. tombstone_ready_job(&pool, &store, id).await; @@ -19052,15 +19113,20 @@ async fn moved_tombstoned_claim_lane(from: i16, to: i16, receipts: bool) { #[tokio::test] async fn test_historical_tombstoned_claim_uses_emitted_lane_low_to_high() { - moved_tombstoned_claim_lane(2, 4, true).await; + moved_tombstoned_claim_lane(2, 4, true, false).await; } #[tokio::test] async fn test_historical_tombstoned_claim_uses_emitted_lane_high_to_low() { - moved_tombstoned_claim_lane(4, 2, true).await; + moved_tombstoned_claim_lane(4, 2, true, false).await; } #[tokio::test] async fn test_historical_tombstoned_legacy_ambiguous_lanes_roll_back() { - moved_tombstoned_claim_lane(2, 4, false).await; + moved_tombstoned_claim_lane(2, 4, false, false).await; +} + +#[tokio::test] +async fn test_historical_tombstoned_receipt_materialized_then_reopened_legacy_uses_evidence() { + moved_tombstoned_claim_lane(2, 4, true, true).await; } #[tokio::test] @@ -19123,7 +19189,7 @@ async fn test_historical_tombstoned_legacy_claim_cancels_body() { #[tokio::test] async fn test_physical_aging_waits_for_claim_head_then_uses_fresh_snapshot() { let (_guard, pool, store, claim, mut claim_tx) = - claim_before_aging_cursor_advance_in_tx(true, Duration::ZERO).await; + claim_before_aging_cursor_advance_in_tx(true, Duration::ZERO, true).await; let holder: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") .fetch_one(claim_tx.as_mut()) .await @@ -19169,6 +19235,6 @@ async fn test_physical_aging_waits_for_claim_head_then_uses_fresh_snapshot() { "the statement after the head lock must see the committed claim" ); assert_eq!(ready_tombstone_count(&pool, &store).await, 0); - assert_claimed_aging_rescue_body(&pool, &store, &claim).await; + assert_claimed_aging_rescue_body(&pool, &store, &claim, false).await; aging_pool.close().await; } From 79d48285ef862293b72710c811db42dcbf5cb69e Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Sat, 7 Nov 2026 16:52:12 +1300 Subject: [PATCH 3/3] Recover retained active bodies and retire old aging copies --- CHANGELOG.md | 23 +- awa-model/src/queue_storage.rs | 826 ++++++++++++++---------- awa/tests/queue_storage_runtime_test.rs | 552 +++++++++++++++- 3 files changed, 1058 insertions(+), 343 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 001f7b41a..ef5b16492 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -27,12 +27,15 @@ order; binaries can roll back to 0.6.13 with the patch retained. See the run lease, preferring live ready bodies over moved source copies. Admin lists and kind counts also include running receipt claims that have not materialized into leases. -- Physical priority aging waits for in-flight claims and rechecks claim evidence - before moving a ready row, including claims committed before their cursor - advances. Rescue, cancellation and completion can recover retained tombstoned - bodies left by older aging, preserving claim-time priority metadata. Emitted - claim-lane evidence takes precedence over colliding live replacements; those - recovered terminal bodies remain self-contained through retry. +- Physical priority aging excludes claimed rows before its batch limit, waits + for in-flight claims and rechecks claim evidence before moving a ready row, + including claims committed before their cursor advances. Active point and + callback reads recover retained tombstoned bodies left by older aging. + Rescue, cancellation and completion by ID recover those bodies and retire + unclaimed copies of the same attempt. Emitted claim-lane evidence takes + precedence over colliding live replacements; those recovered terminal bodies + remain self-contained through retry. Cancellation prefers an active attempt + over its historical available replacement. - Failed terminal writes preserve the ready body's arguments, attempt limit and physical lane. Rust retry and retention carry recover older narrow terminal bodies while their backing remains retained. Stored terminal metadata wins @@ -82,6 +85,14 @@ order; binaries can roll back to 0.6.13 with the patch retained. See the original physical lane if an old aging race left a colliding live replacement. Multiple equally supported bodies cause an explicit ambiguity error and transaction rollback. +- Repair of an old aging replacement runs when a transition hydrates a retained + tombstoned body. Worker completion using its already carried job body does not + perform that repair, and can leave an old replacement claimable. The aging + fence prevents new replacements from updated callers; it cannot fence an + older process explicitly calling physical aging during a rolling upgrade. + Repair takes claim-head row locks without waiting and rolls back on contention; + the operation must retry. It adds weak relation locks and tombstone writes + only on this recovery path. Fresh rechecks rely on READ COMMITTED isolation. ## [0.6.13] — 2026-11-06 diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index 3a7ba35ad..b1a5cd47a 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -6,7 +6,7 @@ use crate::{InsertParams, JobRow, JobState}; use chrono::TimeDelta; use chrono::{DateTime, Utc}; use sqlx::Acquire; -use sqlx::{PgPool, Postgres}; +use sqlx::{FromRow, PgPool, Postgres, Row}; use std::collections::hash_map::DefaultHasher; use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use std::hash::{Hash, Hasher}; @@ -1739,8 +1739,8 @@ impl ReadyJobRow { } } -#[derive(Debug, sqlx::FromRow)] -struct ReadyAgingCandidate { +#[derive(Debug, PartialEq, Eq, sqlx::FromRow)] +struct ReadyRowIdentity { ready_slot: i32, ready_generation: i64, queue: String, @@ -2373,6 +2373,53 @@ struct ReadySnapshotRow { live_body: bool, } +type ReadyBodyKey = (i32, i64, String, i16, i64, i64, i64); +type ReadyBodyLaneKey = (i32, i64, String, i16, i64, i64, i64, i16); + +fn select_claimed_ready_body<'a>( + candidates: &'a [ReadySnapshotRow], + claimed_lanes: &BTreeSet, + reference: &DeletedLeaseRow, +) -> Result<(&'a ReadySnapshotRow, bool), AwaError> { + // Emitted physical-lane evidence wins over a live replacement. + // Without it, retain the live-body compatibility fallback; it + // cannot identify a legacy claim's original colliding lane. + let emitted = |row: &ReadySnapshotRow| { + claimed_lanes.contains(&( + row.ready_slot, + row.ready_generation, + row.queue.clone(), + row.enqueue_shard, + row.lane_seq, + row.job_id, + row.run_lease + 1, + row.priority, + )) + }; + let has_emitted = candidates.iter().any(&emitted); + let rank = |row: &ReadySnapshotRow| { + if has_emitted { + emitted(row) + } else { + row.live_body + } + }; + let strongest = candidates + .iter() + .map(&rank) + .max() + .expect("ready body group is nonempty"); + 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(AwaError::Validation(format!( + "queue storage ready body ambiguous for deleted lease job {} run_lease {}", + reference.job_id, reference.run_lease + ))); + } + Ok((ready, has_emitted)) +} + #[derive(Debug, Clone, sqlx::FromRow)] struct AttemptStateRow { job_id: i64, @@ -8421,6 +8468,72 @@ impl QueueStorage { Ok(newest) } + async fn cancel_materialized_job_lease_tx<'a>( + &self, + tx: &mut sqlx::Transaction<'a, sqlx::Postgres>, + job_id: i64, + ) -> Result, AwaError> { + let schema = self.schema(); + let deleted_lease: Vec = sqlx::query_as(&format!( + r#" + DELETE FROM {schema}.leases + WHERE job_id = $1 + AND state IN ('running', 'waiting_external') + RETURNING + ready_slot, + ready_generation, + job_id, + queue, + state, + priority, + attempt, + run_lease, + max_attempts, + lane_seq, + enqueue_shard, + heartbeat_at, + deadline_at, + attempted_at, + callback_id, + callback_timeout_at + "# + )) + .bind(job_id) + .fetch_all(tx.as_mut()) + .await + .map_err(map_sqlx_error)?; + + if !deleted_lease.is_empty() { + let lease = self + .hydrate_deleted_leases_tx(tx, deleted_lease) + .await? + .into_iter() + .next() + .expect("deleted running lease"); + let done_payload = + Self::payload_with_attempt_state(lease.payload.clone(), lease.progress.clone())?; + let done = lease + .clone() + .into_done_row(JobState::Cancelled, Utc::now(), done_payload); + self.insert_done_rows_tx(tx, std::slice::from_ref(&done), Some(lease.state)) + .await?; + // Receipt-plane consistency: close any matching open + // receipt so the ADR-023 anti-join no longer considers this + // attempt live, and rescue doesn't try to revive it. + self.close_receipt_tx(tx, lease.job_id, lease.run_lease, "cancelled") + .await?; + // Wake any worker currently executing this attempt. + self.notify_cancellation_tx(tx, lease.job_id, lease.run_lease) + .await?; + return Ok(Some(CancelJobTxResult { + row: done.into_job_row()?, + claim_cursor_advance: None, + })); + } + + Ok(None) + } + async fn cancel_job_tx<'a>( &self, tx: &mut sqlx::Transaction<'a, sqlx::Postgres>, @@ -8428,6 +8541,56 @@ impl QueueStorage { ) -> Result, AwaError> { let schema = self.schema(); self.lock_claim_heads_of_jobs_tx(tx, &[job_id]).await?; + if let Some(result) = self.cancel_materialized_job_lease_tx(tx, job_id).await? { + return Ok(Some(result)); + } + + if self.lease_claim_receipts() { + if let Some(run_lease) = self.open_receipt_run_lease_tx(tx, job_id).await? { + self.lock_receipt_attempt_keys_tx(tx, &[(job_id, run_lease)]) + .await?; + let receipt = self + .lock_open_receipt_tx(tx, job_id) + .await? + .filter(|receipt| receipt.1 == run_lease); + if receipt.is_some() { + if let Some(lease) = self + .close_open_receipt_claim_tx(tx, job_id, run_lease, "cancelled") + .await? + { + let done_payload = Self::payload_with_attempt_state( + lease.payload.clone(), + lease.progress.clone(), + )?; + let done = + lease.into_done_row(JobState::Cancelled, Utc::now(), done_payload); + self.insert_done_rows_tx( + tx, + std::slice::from_ref(&done), + Some(JobState::Running), + ) + .await?; + sqlx::query(&format!( + "DELETE FROM {schema}.leases WHERE job_id = $1 AND run_lease = $2" + )) + .bind(job_id) + .bind(run_lease) + .execute(tx.as_mut()) + .await + .map_err(map_sqlx_error)?; + self.notify_cancellation_tx(tx, job_id, run_lease).await?; + return Ok(Some(CancelJobTxResult { + row: done.into_job_row()?, + claim_cursor_advance: None, + })); + } + } + } + if let Some(result) = self.cancel_materialized_job_lease_tx(tx, job_id).await? { + return Ok(Some(result)); + } + } + let claimed = self.ready_row_claimed_sql("ready"); let ready: Option = sqlx::query_as(&format!( r#" @@ -8517,263 +8680,6 @@ impl QueueStorage { })); } - let deleted_lease: Vec = sqlx::query_as(&format!( - r#" - DELETE FROM {schema}.leases - WHERE job_id = $1 - AND state IN ('running', 'waiting_external') - RETURNING - ready_slot, - ready_generation, - job_id, - queue, - state, - priority, - attempt, - run_lease, - max_attempts, - lane_seq, - enqueue_shard, - heartbeat_at, - deadline_at, - attempted_at, - callback_id, - callback_timeout_at - "# - )) - .bind(job_id) - .fetch_all(tx.as_mut()) - .await - .map_err(map_sqlx_error)?; - - if !deleted_lease.is_empty() { - let lease = self - .hydrate_deleted_leases_tx(tx, deleted_lease) - .await? - .into_iter() - .next() - .expect("deleted running lease"); - let done_payload = - Self::payload_with_attempt_state(lease.payload.clone(), lease.progress.clone())?; - let done = lease - .clone() - .into_done_row(JobState::Cancelled, Utc::now(), done_payload); - self.insert_done_rows_tx(tx, std::slice::from_ref(&done), Some(lease.state)) - .await?; - // Receipt-plane consistency: close any matching open - // receipt so the ADR-023 anti-join no longer considers this - // attempt live, and rescue doesn't try to revive it. - self.close_receipt_tx(tx, lease.job_id, lease.run_lease, "cancelled") - .await?; - // Wake any worker currently executing this attempt. - self.notify_cancellation_tx(tx, lease.job_id, lease.run_lease) - .await?; - return Ok(Some(CancelJobTxResult { - row: done.into_job_row()?, - claim_cursor_advance: None, - })); - } - - // ADR-023 receipt-only cancel: the job may be running on a - // receipt-backed short path that never materialized a `leases` - // row. Find it by anti-joining lease_claims with - // lease_claim_closures, cancel it by writing a closure and a - // done row, and notify listening workers. - if self.lease_claim_receipts() { - // A fast completion closes an attempt under its completion lock - // without locking claim rows, so a cancel that only locked the - // rows could commit alongside it. Find the open attempt with plain - // reads, take its completion lock, then look again: a completion - // that got there first has committed its closure by then. - let receipt = match self.open_receipt_run_lease_tx(tx, job_id).await? { - Some(run_lease) => { - self.lock_receipt_attempt_keys_tx(tx, &[(job_id, run_lease)]) - .await?; - self.lock_open_receipt_tx(tx, job_id) - .await? - .filter(|receipt| receipt.1 == run_lease) - } - None => None, - }; - - if let Some(( - claim_slot, - run_lease, - ready_slot, - ready_generation, - queue, - priority, - attempt, - max_attempts, - lane_seq, - claimed_at, - receipt_id, - compact_batch, - )) = receipt - { - // Hydrate the ready row so we can synthesize the done - // row with the original args/payload. - let ready_match: Option = sqlx::query_as(&format!( - r#" - SELECT - 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, - COALESCE(payload, '{{}}'::jsonb) AS payload - FROM {schema}.ready_entries - WHERE job_id = $1 - AND ready_slot = $2 - AND ready_generation = $3 - AND queue = $4 - AND lane_seq = $5 - "# - )) - .bind(job_id) - .bind(ready_slot) - .bind(ready_generation) - .bind(&queue) - .bind(lane_seq) - .fetch_optional(tx.as_mut()) - .await - .map_err(map_sqlx_error)?; - - let Some(ready) = ready_match else { - // Shouldn't happen — the claim references a ready - // row. Fall through to the deferred / not-found - // branches. - return Ok(None); - }; - - let done = DoneJobRow { - ready_slot, - ready_generation, - job_id, - kind: ready.kind, - queue: queue.clone(), - args: ready.args, - state: JobState::Cancelled, - priority, - attempt, - run_lease, - max_attempts, - lane_seq, - enqueue_shard: ready.enqueue_shard, - run_at: ready.run_at, - attempted_at: Some(claimed_at), - finalized_at: Utc::now(), - created_at: ready.created_at, - unique_key: ready.unique_key, - unique_states: ready.unique_states, - payload: ready.payload, - }; - self.insert_done_rows_tx(tx, std::slice::from_ref(&done), Some(JobState::Running)) - .await?; - // Write the closure into the same claim partition. A - // compact batch-sourced claim has no lease_claims row, so - // it closes into the batch ledger the queue prune gate - // counts via compact_count; a deadline lease claim keeps - // its explicit closure so the gate balances it against the - // lease_claims row. - if compact_batch { - sqlx::query(&format!( - r#" - INSERT INTO {schema}.lease_claim_closure_batches ( - claim_slot, - ready_slot, - ready_generation, - outcome, - closed_count, - receipt_ids, - receipt_ranges, - closed_at - ) - VALUES ( - $1, - $2, - $3, - 'cancelled', - 1, - ARRAY[$4::bigint], - int8multirange(int8range($4::bigint, $4::bigint + 1, '[)')), - clock_timestamp() - ) - "# - )) - .bind(claim_slot) - .bind(ready_slot) - .bind(ready_generation) - .bind(receipt_id) - .execute(tx.as_mut()) - .await - .map_err(map_sqlx_error)?; - } else { - sqlx::query(&format!( - r#" - WITH inserted AS ( - INSERT INTO {schema}.lease_claim_closures (claim_slot, job_id, run_lease, outcome, closed_at) - VALUES ($1, $2, $3, 'cancelled', clock_timestamp()) - ON CONFLICT (claim_slot, job_id, run_lease) DO NOTHING - RETURNING claim_slot, job_id, run_lease, closed_at - ), - marked AS ( - UPDATE {schema}.lease_claims AS claims - SET closed_at = COALESCE(claims.closed_at, inserted.closed_at) - FROM inserted - WHERE claims.claim_slot = inserted.claim_slot - AND claims.job_id = inserted.job_id - AND claims.run_lease = inserted.run_lease - RETURNING claims.job_id - ) - SELECT count(*) FROM marked - "# - )) - .bind(claim_slot) - .bind(job_id) - .bind(run_lease) - .execute(tx.as_mut()) - .await - .map_err(map_sqlx_error)?; - } - // Defensive: between the leases DELETE at the top of - // this function and the FOR UPDATE on claims above, a - // concurrent `ensure_running_leases_from_receipts_tx` - // can have materialized a `leases` row for this - // (job_id, run_lease). Materialize and we both lock the - // same claim evidence; whichever ran first commits, the - // other replays under the new snapshot. If materialize - // committed first, that lease is now an orphan pointing - // at a job we're about to mark `cancelled`. Sweep it - // defensively. If no race occurred this is a no-op. - sqlx::query(&format!( - "DELETE FROM {schema}.leases WHERE job_id = $1 AND run_lease = $2" - )) - .bind(job_id) - .bind(run_lease) - .execute(tx.as_mut()) - .await - .map_err(map_sqlx_error)?; - self.notify_cancellation_tx(tx, job_id, run_lease).await?; - return Ok(Some(CancelJobTxResult { - row: done.into_job_row()?, - claim_cursor_advance: None, - })); - } - } - let deferred: Option = sqlx::query_as(&format!( r#" DELETE FROM {schema}.deferred_jobs @@ -9287,8 +9193,19 @@ impl QueueStorage { &self, tx: &mut sqlx::Transaction<'a, sqlx::Postgres>, job_ids: &[i64], + ) -> Result<(), AwaError> { + self.lock_claim_heads_of_jobs_with_wait_tx(tx, job_ids, false) + .await + } + + async fn lock_claim_heads_of_jobs_with_wait_tx( + &self, + tx: &mut sqlx::Transaction<'_, Postgres>, + job_ids: &[i64], + nowait: bool, ) -> Result<(), AwaError> { let schema = self.schema(); + let wait_clause = if nowait { "NOWAIT" } else { "" }; sqlx::query(&format!( r#" SELECT 1 @@ -9299,7 +9216,7 @@ impl QueueStorage { WHERE ready.job_id = ANY($1) ) ORDER BY claims.queue COLLATE "C", claims.priority, claims.enqueue_shard - FOR UPDATE OF claims + FOR UPDATE OF claims {wait_clause} "# )) .bind(job_ids) @@ -9841,12 +9758,13 @@ impl QueueStorage { .map_err(|err| AwaError::Validation(format!("invalid aging interval: {err}")))?; let schema = self.schema(); let mut tx = pool.begin().await.map_err(map_sqlx_error)?; + let claimed = self.ready_row_claimed_sql("ready"); // Claims commit before their cursor advance. Select only a bounded // set of identities before taking heads, without locking ready rows in // the reverse order to a claimer. A later statement rechecks these same // identities after every in-flight claim on their lanes has committed. - let candidates: Vec = sqlx::query_as(&format!( + let candidates: Vec = sqlx::query_as(&format!( r#" SELECT ready.ready_slot, ready.ready_generation, ready.queue, ready.priority, ready.enqueue_shard, ready.lane_seq, @@ -9859,6 +9777,7 @@ impl QueueStorage { WHERE ready.lane_seq >= {schema}.sequence_next_value(claims.seq_name) AND ready.priority > 1 AND ready.run_at <= $1 + AND NOT ({claimed}) AND NOT EXISTS ( SELECT 1 FROM {schema}.ready_tombstones AS tomb WHERE tomb.queue = ready.queue @@ -9883,7 +9802,6 @@ impl QueueStorage { } let job_ids: Vec = candidates.iter().map(|row| row.job_id).collect(); self.lock_claim_heads_of_jobs_tx(&mut tx, &job_ids).await?; - let claimed = self.ready_row_claimed_sql("ready"); let moved: Vec = sqlx::query_as(&format!( r#" @@ -10686,15 +10604,17 @@ impl QueueStorage { Ok(updated as usize) } - async fn hydrate_deleted_leases_tx<'a>( + async fn load_ready_snapshot_candidates_tx( &self, - tx: &mut sqlx::Transaction<'a, sqlx::Postgres>, - deleted: Vec, - ) -> Result, AwaError> { - if deleted.is_empty() { - return Ok(Vec::new()); - } - + tx: &mut sqlx::Transaction<'_, Postgres>, + deleted: &[DeletedLeaseRow], + ) -> Result< + ( + BTreeMap>, + BTreeSet, + ), + AwaError, + > { let schema = self.schema(); let ready_slots: Vec = deleted.iter().map(|row| row.ready_slot).collect(); let ready_generations: Vec = deleted.iter().map(|row| row.ready_generation).collect(); @@ -10769,7 +10689,6 @@ impl QueueStorage { .await .map_err(map_sqlx_error)?; - type ReadyBodyKey = (i32, i64, String, i16, i64, i64, i64); let mut ready_map: BTreeMap> = BTreeMap::new(); for row in ready_rows { ready_map @@ -10786,7 +10705,6 @@ impl QueueStorage { .push(row); } - type ReadyBodyLaneKey = (i32, i64, String, i16, i64, i64, i64, i16); let mut claimed_lanes: BTreeSet = BTreeSet::new(); let ambiguous: Vec<&ReadySnapshotRow> = ready_map .values() @@ -10846,6 +10764,24 @@ impl QueueStorage { claimed_lanes.extend(lanes); } + Ok((ready_map, claimed_lanes)) + } + + async fn hydrate_deleted_leases_tx<'a>( + &self, + tx: &mut sqlx::Transaction<'a, sqlx::Postgres>, + deleted: Vec, + ) -> Result, AwaError> { + if deleted.is_empty() { + return Ok(Vec::new()); + } + + let schema = self.schema(); + let job_ids: Vec = deleted.iter().map(|row| row.job_id).collect(); + 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?; + let attempt_rows: Vec = sqlx::query_as(&format!( r#" WITH refs(job_id, run_lease) AS ( @@ -10996,6 +10932,7 @@ impl QueueStorage { .collect(); let mut hydrated = Vec::with_capacity(deleted.len()); + let mut retained_sources = Vec::new(); for deleted_row in deleted { let candidates = ready_map .get(&( @@ -11013,41 +10950,19 @@ impl QueueStorage { deleted_row.job_id, deleted_row.run_lease )) })?; - // Emitted physical-lane evidence wins over a live replacement. - // Without it, retain the live-body compatibility fallback; it - // cannot identify a legacy claim's original colliding lane. - let emitted = |row: &ReadySnapshotRow| { - claimed_lanes.contains(&( - row.ready_slot, - row.ready_generation, - row.queue.clone(), - row.enqueue_shard, - row.lane_seq, - row.job_id, - row.run_lease + 1, - row.priority, - )) - }; - let has_emitted = candidates.iter().any(&emitted); - let rank = |row: &ReadySnapshotRow| { - if has_emitted { - emitted(row) - } else { - row.live_body - } - }; - let strongest = candidates - .iter() - .map(&rank) - .max() - .expect("ready body group is nonempty"); - 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(AwaError::Validation(format!( - "queue storage ready body ambiguous for deleted lease job {} run_lease {}", - deleted_row.job_id, deleted_row.run_lease - ))); + let (ready, has_emitted) = + select_claimed_ready_body(candidates, &claimed_lanes, &deleted_row)?; + if !ready.live_body { + retained_sources.push(ReadyRowIdentity { + ready_slot: ready.ready_slot, + ready_generation: ready.ready_generation, + queue: ready.queue.clone(), + priority: ready.priority, + enqueue_shard: ready.enqueue_shard, + lane_seq: ready.lane_seq, + job_id: ready.job_id, + run_lease: ready.run_lease, + }); } // Claim-time aging adds this provenance to the returned JobRow, // not the immutable ready payload. Recovery must reconstruct the @@ -11104,9 +11019,146 @@ impl QueueStorage { }); } + if !retained_sources.is_empty() { + self.retire_retained_attempt_copies_tx(tx, &retained_sources) + .await?; + } Ok(hydrated) } + async fn retire_retained_attempt_copies_tx( + &self, + tx: &mut sqlx::Transaction<'_, Postgres>, + sources: &[ReadyRowIdentity], + ) -> Result<(), AwaError> { + let schema = self.schema(); + let mut expected = BTreeMap::new(); + for source in sources { + if expected + .insert((source.job_id, source.run_lease), source) + .is_some() + { + return Err(AwaError::Validation(format!( + "queue storage ready body ambiguous for deleted lease job {} run_lease {}", + source.job_id, + source.run_lease + 1 + ))); + } + } + let job_ids: Vec = sources.iter().map(|row| row.job_id).collect(); + let run_leases: Vec = sources.iter().map(|row| row.run_lease).collect(); + // Hydration already holds attempt locks. A contended head must abort + // instead of waiting in the reverse of the claim/mutation lock order. + self.lock_claim_heads_of_jobs_with_wait_tx(tx, &job_ids, true) + .await?; + let copies: Vec = sqlx::query_as(&format!( + r#" + WITH refs(job_id, run_lease) AS ( + SELECT * FROM unnest($1::bigint[], $2::bigint[]) + ) + SELECT ready.ready_slot, ready.ready_generation, ready.queue, ready.priority, + ready.enqueue_shard, ready.lane_seq, ready.job_id, ready.run_lease + FROM refs + JOIN {schema}.ready_entries AS ready + ON ready.job_id = refs.job_id AND ready.run_lease = refs.run_lease + "# + )) + .bind(&job_ids) + .bind(&run_leases) + .fetch_all(tx.as_mut()) + .await + .map_err(map_sqlx_error)?; + let attempt_sources = copies + .iter() + .map(|row| ring_slot_index(row.ready_slot, self.queue_slot_count(), "queue")) + .collect::, _>>()? + .into_iter() + .map(|slot| { + format!( + "SELECT * FROM {}", + ready_claim_attempt_batch_child_name(schema, slot) + ) + }) + .collect::>() + .join(" UNION ALL "); + if copies.is_empty() { + return Err(AwaError::Validation( + "queue storage ready row missing for deleted lease during retirement".to_string(), + )); + } + let slots: Vec = copies.iter().map(|row| row.ready_slot).collect(); + let generations: Vec = copies.iter().map(|row| row.ready_generation).collect(); + let queues: Vec = copies.iter().map(|row| row.queue.clone()).collect(); + let priorities: Vec = copies.iter().map(|row| row.priority).collect(); + let shards: Vec = copies.iter().map(|row| row.enqueue_shard).collect(); + let sequences: Vec = copies.iter().map(|row| row.lane_seq).collect(); + let copy_job_ids: Vec = copies.iter().map(|row| row.job_id).collect(); + let copy_leases: Vec = copies.iter().map(|row| row.run_lease).collect(); + let emitted: Vec = sqlx::query_as(&format!( + r#" + WITH refs(ready_slot, ready_generation, queue, priority, enqueue_shard, + lane_seq, job_id, run_lease) AS ( + SELECT * FROM unnest($1::int[], $2::bigint[], $3::text[], $4::smallint[], + $5::smallint[], $6::bigint[], $7::bigint[], $8::bigint[]) + ) + SELECT * FROM refs + WHERE EXISTS ( + SELECT 1 FROM ({attempt_sources}) AS attempts + WHERE attempts.ready_slot = refs.ready_slot + AND attempts.ready_generation = refs.ready_generation + AND attempts.queue = refs.queue AND attempts.priority = refs.priority + AND attempts.enqueue_shard = refs.enqueue_shard + AND attempts.first_lane_seq <= refs.lane_seq + AND attempts.next_lane_seq > refs.lane_seq + AND attempts.lane_ranges @> int8range(refs.lane_seq, refs.lane_seq + 1, '[)') + ) + "# + )) + .bind(&slots) + .bind(&generations) + .bind(&queues) + .bind(&priorities) + .bind(&shards) + .bind(&sequences) + .bind(©_job_ids) + .bind(©_leases) + .fetch_all(tx.as_mut()) + .await + .map_err(map_sqlx_error)?; + let mut seen = BTreeSet::new(); + for row in &emitted { + let key = (row.job_id, row.run_lease); + if !seen.insert(key) || expected.get(&key).copied() != Some(row) { + return Err(AwaError::Validation(format!( + "queue storage ready body ambiguous for deleted lease job {} run_lease {}", + row.job_id, + row.run_lease + 1 + ))); + } + } + sqlx::query(&format!( + r#" + INSERT INTO {schema}.ready_tombstones ( + ready_slot, ready_generation, queue, priority, enqueue_shard, lane_seq, job_id + ) + SELECT * FROM unnest($1::int[], $2::bigint[], $3::text[], $4::smallint[], + $5::smallint[], $6::bigint[], $7::bigint[]) + ON CONFLICT DO NOTHING + "# + )) + .bind(&slots) + .bind(&generations) + .bind(&queues) + .bind(&priorities) + .bind(&shards) + .bind(&sequences) + .bind(©_job_ids) + .execute(tx.as_mut()) + .await + .map_err(map_sqlx_error)?; + Ok(()) + } + async fn close_open_receipt_claim_tx<'a>( &self, tx: &mut sqlx::Transaction<'a, sqlx::Postgres>, @@ -12235,6 +12287,87 @@ impl QueueStorage { Ok(rescued) } + async fn active_job_read_row_tx( + &self, + tx: &mut sqlx::Transaction<'_, Postgres>, + row: sqlx::postgres::PgRow, + ) -> Result, AwaError> { + let body_job_id: Option = row.try_get("body_job_id").map_err(map_sqlx_error)?; + if body_job_id.is_some() { + return LeaseJobRow::from_row(&row) + .map_err(map_sqlx_error)? + .into_job_row() + .map(Some); + } + let reference = DeletedLeaseRow::from_row(&row).map_err(map_sqlx_error)?; + let (ready_map, claimed_lanes) = self + .load_ready_snapshot_candidates_tx(tx, std::slice::from_ref(&reference)) + .await?; + let key = ( + reference.ready_slot, + reference.ready_generation, + reference.queue.clone(), + reference.enqueue_shard, + reference.lane_seq, + reference.job_id, + reference.run_lease, + ); + let Some(candidates) = ready_map.get(&key) else { + return Ok(None); + }; + let (ready, _) = select_claimed_ready_body(candidates, &claimed_lanes, &reference)?; + let mut payload = ready.payload.clone(); + if reference.priority < ready.priority { + let metadata = payload + .as_object_mut() + .ok_or_else(|| { + AwaError::Validation("queue storage payload must be a JSON object".into()) + })? + .entry("metadata") + .or_insert_with(|| serde_json::json!({})) + .as_object_mut() + .ok_or_else(|| { + AwaError::Validation( + "queue storage payload metadata must be a JSON object".into(), + ) + })?; + metadata + .entry("_awa_original_priority") + .or_insert_with(|| serde_json::Value::from(i64::from(ready.priority))); + } + LeaseJobRow { + job_id: reference.job_id, + kind: ready.kind.clone(), + queue: ready.queue.clone(), + args: ready.args.clone(), + state: reference.state, + priority: reference.priority, + attempt: reference.attempt, + run_lease: reference.run_lease, + max_attempts: reference.max_attempts, + run_at: ready.run_at, + heartbeat_at: row.try_get("heartbeat_at").map_err(map_sqlx_error)?, + deadline_at: row.try_get("deadline_at").map_err(map_sqlx_error)?, + attempted_at: reference.attempted_at, + finalized_at: None, + created_at: ready.created_at, + unique_key: ready.unique_key.clone(), + callback_id: row.try_get("callback_id").map_err(map_sqlx_error)?, + callback_timeout_at: row.try_get("callback_timeout_at").map_err(map_sqlx_error)?, + callback_filter: row.try_get("callback_filter").map_err(map_sqlx_error)?, + callback_on_complete: row + .try_get("callback_on_complete") + .map_err(map_sqlx_error)?, + callback_on_fail: row.try_get("callback_on_fail").map_err(map_sqlx_error)?, + callback_transform: row.try_get("callback_transform").map_err(map_sqlx_error)?, + payload, + progress: row.try_get("progress").map_err(map_sqlx_error)?, + callback_result: row.try_get("callback_result").map_err(map_sqlx_error)?, + } + .into_job_row() + .map(Some) + } + pub async fn load_job(&self, pool: &PgPool, job_id: i64) -> Result, AwaError> { let schema = self.schema(); let closure_rel = format!("{schema}.lease_claim_closures"); @@ -12388,14 +12521,16 @@ impl QueueStorage { .execute(tx.as_mut()) .await .map_err(map_sqlx_error)?; - let lease_rows: Vec = sqlx::query_as(&format!( + let lease_rows = sqlx::query(&format!( r#" SELECT + ready.job_id AS body_job_id, lease.ready_slot, lease.ready_generation, lease.job_id, + lease.enqueue_shard, ready.kind, - ready.queue, + lease.queue, ready.args, lease.state, lease.priority, @@ -12421,7 +12556,7 @@ impl QueueStorage { attempt.progress, attempt.callback_result FROM {schema}.leases AS lease - JOIN ({ready_sources}) AS ready + LEFT JOIN ({ready_sources}) AS ready ON ready.ready_slot = lease.ready_slot AND ready.ready_generation = lease.ready_generation AND ready.queue = lease.queue @@ -12451,15 +12586,18 @@ impl QueueStorage { .fetch_all(tx.as_mut()) .await .map_err(map_sqlx_error)?; + for row in lease_rows { + if let Some(job) = self.active_job_read_row_tx(&mut tx, row).await? { + candidates.push(job); + } + } + sqlx::Executor::execute( tx.as_mut(), "ROLLBACK TO SAVEPOINT awa_read_leg; RELEASE SAVEPOINT awa_read_leg", ) .await .map_err(map_sqlx_error)?; - for row in lease_rows { - candidates.push(row.into_job_row()?); - } // Report receipt-backed attempts as running by anti-joining // lease_claims against every durable closure evidence shape. @@ -12467,14 +12605,16 @@ impl QueueStorage { .execute(tx.as_mut()) .await .map_err(map_sqlx_error)?; - let lease_claim_rows: Vec = sqlx::query_as(&format!( + let lease_claim_rows = sqlx::query(&format!( r#" SELECT + ready.job_id AS body_job_id, claims.ready_slot, claims.ready_generation, claims.job_id, + claims.enqueue_shard, ready.kind, - ready.queue, + claims.queue, ready.args, 'running'::awa.job_state AS state, claims.priority, @@ -12500,7 +12640,7 @@ impl QueueStorage { attempt.progress, attempt.callback_result FROM {schema}.lease_claims AS claims - JOIN ({ready_sources}) AS ready + LEFT JOIN ({ready_sources}) AS ready ON ready.ready_slot = claims.ready_slot AND ready.ready_generation = claims.ready_generation AND ready.queue = claims.queue @@ -12539,15 +12679,18 @@ impl QueueStorage { .fetch_all(tx.as_mut()) .await .map_err(map_sqlx_error)?; + for row in lease_claim_rows { + if let Some(job) = self.active_job_read_row_tx(&mut tx, row).await? { + candidates.push(job); + } + } + sqlx::Executor::execute( tx.as_mut(), "ROLLBACK TO SAVEPOINT awa_read_leg; RELEASE SAVEPOINT awa_read_leg", ) .await .map_err(map_sqlx_error)?; - for row in lease_claim_rows { - candidates.push(row.into_job_row()?); - } // Zero-deadline receipt claims are stored as compact batches. Expand // them only for this admin read, and report still-open items as @@ -12557,14 +12700,16 @@ impl QueueStorage { .execute(tx.as_mut()) .await .map_err(map_sqlx_error)?; - let lease_claim_batch_rows: Vec = sqlx::query_as(&format!( + let lease_claim_batch_rows = sqlx::query(&format!( r#" SELECT + ready.job_id AS body_job_id, claim_batches.ready_slot, claim_batches.ready_generation, items.job_id, + claim_batches.enqueue_shard, ready.kind, - ready.queue, + claim_batches.queue, ready.args, 'running'::awa.job_state AS state, claim_batches.priority, @@ -12598,7 +12743,7 @@ impl QueueStorage { claim_batches.attempts, claim_batches.max_attempts ) AS items(job_id, run_lease, receipt_id, lane_seq, attempt, max_attempts) - JOIN ({ready_sources}) AS ready + LEFT JOIN ({ready_sources}) AS ready ON ready.ready_slot = claim_batches.ready_slot AND ready.ready_generation = claim_batches.ready_generation AND ready.queue = claim_batches.queue @@ -12660,15 +12805,18 @@ impl QueueStorage { .fetch_all(tx.as_mut()) .await .map_err(map_sqlx_error)?; + for row in lease_claim_batch_rows { + if let Some(job) = self.active_job_read_row_tx(&mut tx, row).await? { + candidates.push(job); + } + } + sqlx::Executor::execute( tx.as_mut(), "ROLLBACK TO SAVEPOINT awa_read_leg; RELEASE SAVEPOINT awa_read_leg", ) .await .map_err(map_sqlx_error)?; - for row in lease_claim_batch_rows { - candidates.push(row.into_job_row()?); - } sqlx::query("SAVEPOINT awa_read_leg") .execute(tx.as_mut()) @@ -13051,14 +13199,16 @@ impl QueueStorage { run_lease: i64, ) -> Result, AwaError> { let schema = self.schema(); - let row: Option = sqlx::query_as(&format!( + let row = sqlx::query(&format!( r#" SELECT + ready.job_id AS body_job_id, lease.ready_slot, lease.ready_generation, lease.job_id, + lease.enqueue_shard, ready.kind, - ready.queue, + lease.queue, ready.args, lease.state, lease.priority, @@ -13084,7 +13234,7 @@ impl QueueStorage { attempt.progress, attempt.callback_result FROM {schema}.leases AS lease - JOIN {schema}.ready_entries AS ready + LEFT JOIN {schema}.ready_entries AS ready ON ready.ready_slot = lease.ready_slot AND ready.ready_generation = lease.ready_generation AND ready.queue = lease.queue @@ -13115,7 +13265,10 @@ impl QueueStorage { .fetch_optional(tx.as_mut()) .await .map_err(map_sqlx_error)?; - row.map(LeaseJobRow::into_job_row).transpose() + match row { + Some(row) => self.active_job_read_row_tx(tx, row).await, + None => Ok(None), + } } pub async fn enter_callback_wait( @@ -13259,14 +13412,16 @@ impl QueueStorage { } else { "" }; - let row: Option = sqlx::query_as(&format!( + let row = sqlx::query(&format!( r#" SELECT + ready.job_id AS body_job_id, lease.ready_slot, lease.ready_generation, lease.job_id, + lease.enqueue_shard, ready.kind, - ready.queue, + lease.queue, ready.args, lease.state, lease.priority, @@ -13292,7 +13447,7 @@ impl QueueStorage { attempt.progress, attempt.callback_result FROM {} AS lease - JOIN {schema}.ready_entries AS ready + LEFT JOIN {schema}.ready_entries AS ready ON ready.ready_slot = lease.ready_slot AND ready.ready_generation = lease.ready_generation AND ready.queue = lease.queue @@ -13330,7 +13485,10 @@ impl QueueStorage { .await .map_err(map_sqlx_error)?; - row.map(LeaseJobRow::into_job_row).transpose() + match row { + Some(row) => self.active_job_read_row_tx(tx, row).await, + None => Ok(None), + } } #[tracing::instrument(skip(self, pool, payload), name = "queue_storage.complete_external")] diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index 4ac81b712..2e025a863 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -18873,6 +18873,13 @@ async fn assert_claimed_aging_rescue_body( ), "only conflicting historical body recovery needs a self-contained terminal row" ); + for _ in 0..8 { + let available = store + .claim_runtime_batch(pool, "qs_claimed_aging_body", 1, Duration::from_secs(300)) + .await + .expect("drain ready prefixes before explicit retry"); + assert!(available.iter().all(|row| row.job.id != claim.job_id)); + } let retried = store .retry_job(pool, claim.job_id) .await @@ -18882,6 +18889,19 @@ async fn assert_claimed_aging_rescue_body( assert_eq!(retried.max_attempts, 1); assert_eq!(retried.priority, 4); assert_eq!(retried.metadata["_awa_original_priority"], 4); + let mut claimed_retries = Vec::new(); + for _ in 0..8 { + claimed_retries.extend( + store + .claim_runtime_batch(pool, "qs_claimed_aging_body", 1, Duration::from_secs(300)) + .await + .expect("claim only the new attempt after explicit retry") + .into_iter() + .filter(|row| row.job.id == claim.job_id), + ); + } + assert_eq!(claimed_retries.len(), 1); + assert_eq!(claimed_retries[0].job.run_lease, claim.run_lease + 1); } async fn aging_skips_committed_claim_before_cursor(receipts: bool, deadline: Duration) { @@ -18898,12 +18918,12 @@ async fn aging_skips_committed_claim_before_cursor(receipts: bool, deadline: Dur assert_claimed_aging_rescue_body(&pool, &store, &claim, false).await; } -async fn historical_tombstoned_claim_hydrates( +async fn historical_tombstoned_claim_fixture( receipts: bool, deadline: Duration, colliding_position: bool, -) { - let (_guard, pool, store, claim, tx) = +) -> (TestDbGuard, sqlx::PgPool, QueueStorage, RawReceiptClaimRow) { + let (guard, pool, store, claim, tx) = claim_before_aging_cursor_advance_in_tx(receipts, deadline, !colliding_position).await; tx.commit() .await @@ -18933,6 +18953,16 @@ async fn historical_tombstoned_claim_hydrates( || jsonb_build_object('_awa_original_priority',priority)) FROM {schema}.ready_entries WHERE job_id=$1" )).bind(claim.job_id).execute(tx.as_mut()).await.expect("historical replacement body"); + sqlx::query(&format!( + "INSERT INTO {schema}.ready_segments + (ready_slot,ready_generation,queue,priority,enqueue_shard,first_lane_seq,next_lane_seq,first_run_at) + SELECT ready_slot,ready_generation,queue,priority,enqueue_shard,lane_seq,lane_seq+1,run_at + FROM {schema}.ready_entries WHERE job_id=$1 AND priority=3" + )) + .bind(claim.job_id) + .execute(tx.as_mut()) + .await + .expect("publish historical replacement segment as physical aging does"); tx.commit() .await .expect("commit historical source and replacement"); @@ -18957,6 +18987,16 @@ async fn historical_tombstoned_claim_hydrates( vec![(4, None), (3, Some(4))], "old physical aging adds provenance only to its replacement body" ); + (guard, pool, store, claim) +} + +async fn historical_tombstoned_claim_hydrates( + receipts: bool, + deadline: Duration, + colliding_position: bool, +) { + let (_guard, pool, store, claim) = + historical_tombstoned_claim_fixture(receipts, deadline, colliding_position).await; assert_claimed_aging_rescue_body(&pool, &store, &claim, colliding_position).await; } @@ -19238,3 +19278,509 @@ async fn test_physical_aging_waits_for_claim_head_then_uses_fresh_snapshot() { assert_claimed_aging_rescue_body(&pool, &store, &claim, false).await; aging_pool.close().await; } + +async fn aging_skips_claimed_candidates_before_limit(receipts: bool, deadline: Duration) { + let (_guard, pool, store, claim) = claim_before_aging_cursor_advance(receipts, deadline).await; + let waiting = enqueue_job( + &pool, + &store, + &CompleteJob { id: 9044 }, + InsertOpts { + queue: "qs_claimed_aging_body".into(), + priority: 4, + ..Default::default() + }, + ) + .await; + backdate_one_lane_ready_job_run_at( + &pool, + store.schema(), + waiting, + Utc::now() - chrono::Duration::minutes(4), + ) + .await; + let aged = store + .age_waiting_priorities(&pool, Duration::from_secs(60), 1) + .await + .expect("age past the oldest recorded claim"); + assert_eq!( + aged, + vec![waiting], + "claimed rows cannot consume the candidate limit" + ); + let waiting = store.load_job(&pool, waiting).await.unwrap().unwrap(); + assert_eq!(waiting.priority, 3); + assert_eq!(waiting.metadata["_awa_original_priority"], 4); + assert_claimed_aging_rescue_body(&pool, &store, &claim, false).await; +} + +#[tokio::test] +async fn test_physical_aging_skips_claimed_candidates_before_limit_compact() { + aging_skips_claimed_candidates_before_limit(true, Duration::ZERO).await; +} +#[tokio::test] +async fn test_physical_aging_skips_claimed_candidates_before_limit_deadline() { + aging_skips_claimed_candidates_before_limit(true, Duration::from_secs(300)).await; +} +#[tokio::test] +async fn test_physical_aging_skips_claimed_candidates_before_limit_legacy() { + aging_skips_claimed_candidates_before_limit(false, Duration::from_secs(300)).await; +} + +fn assert_historical_active_body(job: &JobRow, claim: &RawReceiptClaimRow, state: JobState) { + assert_eq!(job.id, claim.job_id); + assert_eq!(job.state, state); + assert_eq!(job.run_lease, claim.run_lease); + assert_eq!(job.priority, claim.priority); + assert_eq!(job.attempt, claim.attempt); + assert_eq!(job.args, serde_json::json!({"id":9041})); + assert_eq!(job.max_attempts, 1); + assert_eq!(job.metadata["retained"], "claimed-before-aging"); + assert_eq!(job.metadata["_awa_original_priority"], 4); +} + +async fn historical_tombstoned_claim_point_read(receipts: bool, deadline: Duration) { + let (_guard, pool, store, claim) = + historical_tombstoned_claim_fixture(receipts, deadline, false).await; + let loaded = store + .load_job(&pool, claim.job_id) + .await + .expect("read historically tombstoned active claim") + .expect("claimed job remains visible"); + assert_historical_active_body(&loaded, &claim, JobState::Running); + assert_claimed_aging_rescue_body(&pool, &store, &claim, false).await; +} + +#[tokio::test] +async fn test_historical_tombstoned_compact_claim_point_read_is_running() { + historical_tombstoned_claim_point_read(true, Duration::ZERO).await; +} + +#[tokio::test] +async fn test_historical_tombstoned_deadline_claim_point_read_is_running() { + historical_tombstoned_claim_point_read(true, Duration::from_secs(300)).await; +} + +#[tokio::test] +async fn test_historical_tombstoned_legacy_lease_point_read_is_running() { + historical_tombstoned_claim_point_read(false, Duration::from_secs(300)).await; +} + +async fn historical_tombstoned_claim_callback_reads(receipts: bool, deadline: Duration) { + let (_guard, pool, store, claim) = + historical_tombstoned_claim_fixture(receipts, deadline, false).await; + let callback_id = store + .register_callback( + &pool, + claim.job_id, + claim.run_lease, + Duration::from_secs(300), + ) + .await + .expect("register callback for retained claim"); + let schema = store.schema(); + let progress = serde_json::json!({"phase":"retained-body"}); + sqlx::query(&format!( + "INSERT INTO {schema}.attempt_state + (job_id,run_lease,progress,callback_filter,callback_on_complete,callback_on_fail,callback_transform,updated_at) + VALUES ($1,$2,$3,'payload.ok','payload.done','payload.fail','payload.value',clock_timestamp()) + ON CONFLICT (job_id,run_lease) DO UPDATE SET + progress=EXCLUDED.progress,callback_filter=EXCLUDED.callback_filter, + callback_on_complete=EXCLUDED.callback_on_complete,callback_on_fail=EXCLUDED.callback_on_fail, + callback_transform=EXCLUDED.callback_transform" + )).bind(claim.job_id).bind(claim.run_lease).bind(&progress) + .execute(&pool).await.expect("set mutable callback state"); + let mut tx = pool.begin().await.unwrap(); + let running = store + .load_active_lease_in_tx(&mut tx, claim.job_id, claim.run_lease) + .await + .expect("raw active lease lookup") + .expect("running lease still has retained body"); + assert_historical_active_body(&running, &claim, JobState::Running); + assert_eq!(running.callback_id, Some(callback_id)); + assert_eq!(running.progress, Some(progress.clone())); + assert!(store + .enter_callback_wait_in_tx(&mut tx, claim.job_id, claim.run_lease, callback_id) + .await + .expect("park retained claim")); + let parked = store + .load_active_lease_in_tx(&mut tx, claim.job_id, claim.run_lease) + .await + .expect("read post-park row in caller transaction") + .expect("parked body"); + assert_historical_active_body(&parked, &claim, JobState::WaitingExternal); + assert_eq!(parked.callback_id, Some(callback_id)); + assert_eq!(parked.callback_timeout_at, running.callback_timeout_at); + assert_eq!(parked.heartbeat_at, None); + assert_eq!(parked.deadline_at, None); + for for_update in [false, true] { + let callback = store + .callback_job_in_tx(&mut tx, callback_id, Some(claim.run_lease), for_update) + .await + .expect("lookup retained callback body") + .expect("callback exists"); + assert_historical_active_body(&callback, &claim, JobState::WaitingExternal); + assert_eq!(callback.callback_id, Some(callback_id)); + assert_eq!(callback.callback_timeout_at, parked.callback_timeout_at); + assert_eq!(callback.progress, Some(progress.clone())); + assert_eq!(callback.callback_filter.as_deref(), Some("payload.ok")); + assert_eq!( + callback.callback_on_complete.as_deref(), + Some("payload.done") + ); + assert_eq!(callback.callback_on_fail.as_deref(), Some("payload.fail")); + assert_eq!( + callback.callback_transform.as_deref(), + Some("payload.value") + ); + } + tx.commit().await.expect("commit parked lease"); + let loaded = store + .load_job(&pool, claim.job_id) + .await + .expect("point read parked lease") + .unwrap(); + assert_historical_active_body(&loaded, &claim, JobState::WaitingExternal); + let mut locked = pool.begin().await.unwrap(); + store + .callback_job_in_tx(&mut locked, callback_id, None, true) + .await + .expect("lock callback lease despite missing live body") + .unwrap(); + let mut contender = pool.begin().await.unwrap(); + let err = sqlx::query(&format!( + "SELECT job_id FROM {schema}.leases WHERE job_id=$1 AND run_lease=$2 FOR UPDATE NOWAIT" + )) + .bind(claim.job_id) + .bind(claim.run_lease) + .fetch_one(contender.as_mut()) + .await + .expect_err("callback lookup must retain its lease row lock"); + assert_eq!( + err.as_database_error() + .and_then(|err| err.code()) + .as_deref(), + Some("55P03") + ); + contender.rollback().await.unwrap(); + locked.rollback().await.unwrap(); + let resumed = store + .complete_external( + &pool, + callback_id, + Some(serde_json::json!({"ok":true})), + Some(claim.run_lease), + true, + ) + .await + .expect("resume callback and read retained body inside update transaction"); + assert_historical_active_body(&resumed, &claim, JobState::Running); + assert_eq!(resumed.callback_id, None); + assert_eq!(resumed.progress, Some(progress)); + assert_eq!( + resumed.metadata["_awa_callback_result"], + serde_json::json!({"ok":true}) + ); +} + +#[tokio::test] +async fn test_historical_tombstoned_compact_claim_callback_reads_and_resume() { + historical_tombstoned_claim_callback_reads(true, Duration::ZERO).await; +} + +#[tokio::test] +async fn test_historical_tombstoned_deadline_claim_callback_reads_and_resume() { + historical_tombstoned_claim_callback_reads(true, Duration::from_secs(300)).await; +} + +#[tokio::test] +async fn test_historical_tombstoned_legacy_lease_callback_reads_and_resume() { + historical_tombstoned_claim_callback_reads(false, Duration::from_secs(300)).await; +} + +#[tokio::test] +async fn test_historical_tombstoned_colliding_receipt_point_reads_use_emitted_lane() { + let (_guard, pool, store, claim) = + historical_tombstoned_claim_fixture(true, Duration::ZERO, true).await; + tombstone_ready_job(&pool, &store, claim.job_id).await; + let loaded = store + .load_job(&pool, claim.job_id) + .await + .expect("read emitted claimed lane") + .unwrap(); + assert_historical_active_body(&loaded, &claim, JobState::Running); + let callback_id = store + .register_callback( + &pool, + claim.job_id, + claim.run_lease, + Duration::from_secs(300), + ) + .await + .expect("materialize retained emitted claim"); + let mut tx = pool.begin().await.unwrap(); + let active = store + .load_active_lease_in_tx(&mut tx, claim.job_id, claim.run_lease) + .await + .expect("load retained emitted lease") + .unwrap(); + assert_historical_active_body(&active, &claim, JobState::Running); + let callback = store + .callback_job_in_tx(&mut tx, callback_id, None, true) + .await + .expect("callback resolves emitted lane") + .unwrap(); + assert_historical_active_body(&callback, &claim, JobState::Running); + tx.rollback().await.unwrap(); +} + +#[tokio::test] +async fn test_historical_tombstoned_colliding_legacy_point_reads_fail_ambiguous() { + let (_guard, pool, store, claim) = + historical_tombstoned_claim_fixture(false, Duration::from_secs(300), true).await; + let callback_id = store + .register_callback( + &pool, + claim.job_id, + claim.run_lease, + Duration::from_secs(300), + ) + .await + .expect("register legacy callback"); + tombstone_ready_job(&pool, &store, claim.job_id).await; + let expected = format!( + "queue storage ready body ambiguous for deleted lease job {} run_lease {}", + claim.job_id, claim.run_lease + ); + let err = store + .load_job(&pool, claim.job_id) + .await + .expect_err("point read cannot choose legacy lane"); + assert!( + matches!(err, AwaError::Validation(ref message) if message == &expected), + "{err}" + ); + let mut tx = pool.begin().await.unwrap(); + let err = store + .load_active_lease_in_tx(&mut tx, claim.job_id, claim.run_lease) + .await + .expect_err("active lookup cannot choose legacy lane"); + assert!( + matches!(err, AwaError::Validation(ref message) if message == &expected), + "{err}" + ); + let err = store + .callback_job_in_tx(&mut tx, callback_id, None, true) + .await + .expect_err("callback lookup cannot choose legacy lane"); + assert!( + matches!(err, AwaError::Validation(ref message) if message == &expected), + "{err}" + ); + tx.rollback().await.unwrap(); + assert_eq!(lease_count(&pool, &store).await, 1); +} + +async fn historical_generic_cancel_live_replacement(deadline: Duration, colliding_position: bool) { + let (_guard, pool, store, claim) = + historical_tombstoned_claim_fixture(true, deadline, colliding_position).await; + let schema = store.schema(); + let live_bodies_sql = format!( + "SELECT count(*)::bigint FROM {schema}.ready_entries AS ready + WHERE ready.job_id=$1 AND NOT EXISTS ( + SELECT 1 FROM {schema}.ready_tombstones AS tomb + WHERE tomb.ready_slot=ready.ready_slot AND tomb.ready_generation=ready.ready_generation + AND tomb.queue=ready.queue AND tomb.priority=ready.priority + AND tomb.enqueue_shard=ready.enqueue_shard AND tomb.lane_seq=ready.lane_seq + AND tomb.job_id=ready.job_id)" + ); + let live_before: i64 = sqlx::query_scalar(&live_bodies_sql) + .bind(claim.job_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(live_before, 1); + let cancelled = store + .cancel_job(&pool, claim.job_id) + .await + .expect("cancel running receipt ahead of its historical live replacement") + .expect("claimed attempt is cancelled"); + assert_eq!(cancelled.id, claim.job_id); + assert_eq!(cancelled.state, JobState::Cancelled); + assert_eq!(cancelled.run_lease, claim.run_lease); + assert_eq!(cancelled.args, serde_json::json!({"id":9041})); + assert_eq!(cancelled.max_attempts, 1); + assert_eq!(cancelled.priority, 4); + assert_eq!(cancelled.metadata["retained"], "claimed-before-aging"); + assert_eq!(cancelled.metadata["_awa_original_priority"], 4); + let live_after: i64 = sqlx::query_scalar(&live_bodies_sql) + .bind(claim.job_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(live_after, 0); + let loaded = store + .load_job(&pool, claim.job_id) + .await + .expect("read cancelled receipt") + .expect("cancelled receipt remains visible"); + assert_eq!(loaded.state, JobState::Cancelled); + assert_eq!(loaded.run_lease, claim.run_lease); + assert_eq!(loaded.args, cancelled.args); + assert_eq!(loaded.priority, 4); + assert_eq!(loaded.metadata["_awa_original_priority"], 4); + assert!(store + .rescue_stale_heartbeats(&pool, Duration::ZERO) + .await + .expect("cancelled receipt is closed") + .is_empty()); + for _ in 0..8 { + let before_retry = store + .claim_runtime_batch(&pool, "qs_claimed_aging_body", 1, deadline) + .await + .expect("claim any unrelated ready job after cancel"); + assert!(before_retry + .iter() + .all(|entry| entry.job.id != claim.job_id)); + } + let retried = store + .retry_job(&pool, claim.job_id) + .await + .expect("retry generic receipt cancellation") + .expect("cancelled receipt retries"); + assert_eq!(retried.args, cancelled.args); + assert_eq!(retried.max_attempts, 1); + assert_eq!(retried.priority, 4); + assert_eq!(retried.metadata["_awa_original_priority"], 4); + let mut claimed_retry = Vec::new(); + for _ in 0..8 { + claimed_retry.extend( + store + .claim_runtime_batch(&pool, "qs_claimed_aging_body", 1, deadline) + .await + .expect("claim the explicit retry"), + ); + } + assert_eq!(claimed_retry.len(), 1); + assert_eq!(claimed_retry[0].job.id, claim.job_id); + assert_eq!(claimed_retry[0].job.run_lease, claim.run_lease + 1); + assert_eq!(claimed_retry[0].job.priority, 4); + assert!(store + .claim_runtime_batch(&pool, "qs_claimed_aging_body", 10, deadline) + .await + .expect("old historical copy cannot emit another claim") + .is_empty()); +} + +#[tokio::test] +async fn test_generic_cancel_historical_colliding_live_compact_replacement_keeps_claimed_lane() { + historical_generic_cancel_live_replacement(Duration::ZERO, true).await; +} + +#[tokio::test] +async fn test_generic_cancel_historical_live_deadline_replacement_keeps_claimed_lane() { + historical_generic_cancel_live_replacement(Duration::from_secs(300), false).await; +} + +async fn historical_recovery_head_contention_rolls_back(receipts: bool, deadline: Duration) { + let (_guard, pool, store, claim) = + historical_tombstoned_claim_fixture(receipts, deadline, false).await; + let schema = store.schema(); + let mut blocker = pool.begin().await.unwrap(); + sqlx::query(&format!( + "SELECT 1 FROM {schema}.queue_claim_heads + WHERE queue='qs_claimed_aging_body' AND priority=3 AND enqueue_shard=0 FOR UPDATE" + )) + .execute(blocker.as_mut()) + .await + .unwrap(); + let result = tokio::time::timeout( + scaled_timeout(Duration::from_secs(2)), + store.rescue_stale_heartbeats(&pool, Duration::ZERO), + ) + .await + .expect("contended recovery head fails without waiting"); + assert!( + matches!(result, Err(AwaError::Database(ref error)) if error.as_database_error().is_some_and(|error| error.code().as_deref() == Some("55P03"))), + "recovery must fail with lock_not_available: {result:?}" + ); + let done: i64 = sqlx::query_scalar(&format!( + "SELECT count(*) FROM {schema}.done_entries WHERE job_id=$1" + )) + .bind(claim.job_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(done, 0); + assert_eq!(ready_tombstone_count(&pool, &store).await, 1); + let active = store.load_job(&pool, claim.job_id).await.unwrap().unwrap(); + assert_historical_active_body(&active, &claim, JobState::Running); + blocker.rollback().await.unwrap(); + assert_claimed_aging_rescue_body(&pool, &store, &claim, false).await; +} + +#[tokio::test] +async fn test_historical_compact_recovery_head_contention_rolls_back() { + historical_recovery_head_contention_rolls_back(true, Duration::ZERO).await; +} + +#[tokio::test] +async fn test_historical_legacy_recovery_head_contention_rolls_back() { + historical_recovery_head_contention_rolls_back(false, Duration::from_secs(300)).await; +} + +async fn historical_tombstoned_claim_completes(receipts: bool, deadline: Duration) { + let (_guard, pool, store, claim) = + historical_tombstoned_claim_fixture(receipts, deadline, false).await; + if receipts { + store + .register_callback( + &pool, + claim.job_id, + claim.run_lease, + Duration::from_secs(300), + ) + .await + .expect("materialize retained claim before completion"); + } + let completed = store + .complete_job_batch_by_id(&pool, &[(claim.job_id, claim.run_lease)]) + .await + .expect("complete retained body after old aging race"); + assert_eq!(completed, vec![(claim.job_id, claim.run_lease)]); + let loaded = store.load_job(&pool, claim.job_id).await.unwrap().unwrap(); + assert_eq!(loaded.state, JobState::Completed); + assert_eq!(loaded.run_lease, claim.run_lease); + assert_eq!(loaded.args, serde_json::json!({"id":9041})); + assert_eq!(loaded.max_attempts, 1); + assert_eq!(loaded.priority, 4); + assert_eq!(loaded.metadata["_awa_original_priority"], 4); + for _ in 0..8 { + let available = store + .claim_runtime_batch(&pool, "qs_claimed_aging_body", 10, deadline) + .await + .expect("old aging replacement cannot claim after hydrated completion"); + assert!(available.iter().all(|row| row.job.id != claim.job_id)); + } + assert!(store + .rescue_stale_heartbeats(&pool, Duration::ZERO) + .await + .unwrap() + .iter() + .all(|row| row.id != claim.job_id)); +} + +#[tokio::test] +async fn test_historical_tombstoned_compact_claim_completes_and_retires_copy() { + historical_tombstoned_claim_completes(true, Duration::ZERO).await; +} + +#[tokio::test] +async fn test_historical_tombstoned_deadline_claim_completes_and_retires_copy() { + historical_tombstoned_claim_completes(true, Duration::from_secs(300)).await; +} + +#[tokio::test] +async fn test_historical_tombstoned_legacy_claim_completes_and_retires_copy() { + historical_tombstoned_claim_completes(false, Duration::from_secs(300)).await; +}