From 324cea555c7f2553b53fcb1cf0faafa367e5ce05 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Thu, 12 Nov 2026 08:01:30 +0000 Subject: [PATCH 1/4] fix(queue-storage): make an explicit progress flush wait for the attempt lock A heartbeat batch leaves out receipt attempts whose lock another transaction holds. flush_progress shared that helper, so a flush that overlapped a heartbeat or a completion of the same attempt wrote nothing and still returned Ok. The flush now waits for the lock; only heartbeat batches skip. --- awa-model/src/queue_storage.rs | 28 ++++++-- awa/tests/queue_storage_runtime_test.rs | 91 +++++++++++++++++++++++++ 2 files changed, 112 insertions(+), 7 deletions(-) diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index 2695177a7..ed1289303 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -11214,10 +11214,14 @@ impl QueueStorage { Ok(updated as usize) } + /// With `skip_locked`, attempts whose lock another transaction holds are + /// left out, as a heartbeat batch wants. Without it the call waits for + /// each attempt's lock, so an explicit flush is never dropped. async fn upsert_attempt_state_progress_from_receipts_tx<'a>( &self, tx: &mut sqlx::Transaction<'a, sqlx::Postgres>, jobs: &[(i64, i64, serde_json::Value)], + skip_locked: bool, ) -> Result { if jobs.is_empty() { return Ok(0); @@ -11227,14 +11231,23 @@ impl QueueStorage { .iter() .map(|(job_id, run_lease, _)| (*job_id, *run_lease)) .collect(); - let locked: HashSet<_> = self - .try_lock_receipt_attempt_keys_tx(tx, &receipt_jobs) - .await? - .into_iter() - .collect(); + let locked: Option> = if skip_locked { + Some( + self.try_lock_receipt_attempt_keys_tx(tx, &receipt_jobs) + .await? + .into_iter() + .collect(), + ) + } else { + None + }; let jobs: Vec<_> = jobs .iter() - .filter(|(job_id, run_lease, _)| locked.contains(&(*job_id, *run_lease))) + .filter(|(job_id, run_lease, _)| { + locked + .as_ref() + .is_none_or(|locked| locked.contains(&(*job_id, *run_lease))) + }) .collect(); if jobs.is_empty() { return Ok(0); @@ -15069,6 +15082,7 @@ impl QueueStorage { self.upsert_attempt_state_progress_from_receipts_tx( &mut tx, &[(job_id, run_lease, progress.clone())], + false, ) .await?; } @@ -15312,7 +15326,7 @@ impl QueueStorage { // anti-joined against durable closure evidence so it picks // up every open claim regardless of whether // `materialize_claims` has fanned it out to `leases` yet. - self.upsert_attempt_state_progress_from_receipts_tx(&mut tx, jobs) + self.upsert_attempt_state_progress_from_receipts_tx(&mut tx, jobs, true) .await? } else { // Prelock independently so a finalizer cannot stall the entire diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index ea033c9f4..48c8c5c4b 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -6623,6 +6623,97 @@ async fn test_queue_storage_receipt_rescue_cursor_sweeps_past_fresh_claims_and_w assert_eq!(open_receipt_claim_count(&pool, &store).await, 0); } +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_queue_storage_flush_progress_waits_for_a_locked_receipt_attempt() { + let (_db_guard, pool) = setup_pool(10).await; + let queue = "qs_flush_progress_locked"; + let schema = "awa_qs_flush_progress_locked"; + let store = Arc::new( + create_store_with_config( + &pool, + QueueStorageConfig { + schema: schema.to_string(), + queue_slot_count: 4, + lease_slot_count: 2, + queue_stripe_count: 1, + lease_claim_receipts: true, + claim_slot_count: 2, + }, + ) + .await, + ); + + let job_id = enqueue_job( + &pool, + &store, + &CompleteJob { id: 78 }, + InsertOpts { + queue: queue.to_string(), + ..Default::default() + }, + ) + .await; + let claim = store + .claim_runtime_batch(&pool, queue, 1, Duration::ZERO) + .await + .expect("claim receipt-backed job") + .into_iter() + .next() + .expect("claimed job"); + assert!(claim.claim.lease_claim_receipt); + let run_lease = claim.job.run_lease; + + let mut attempt_lock = pool.begin().await.expect("begin attempt lock tx"); + sqlx::query( + r#" + SELECT pg_catalog.pg_advisory_xact_lock( + pg_catalog.hashtextextended( + format('awa.receipt.complete:%s:%s', $1::bigint, $2::bigint), + 0 + ) + ) + "#, + ) + .bind(job_id) + .bind(run_lease) + .execute(attempt_lock.as_mut()) + .await + .expect("hold the attempt lock"); + + let flush = { + let store = store.clone(); + let pool = pool.clone(); + tokio::spawn(async move { + store + .flush_progress(&pool, job_id, run_lease, serde_json::json!({"step": 3})) + .await + }) + }; + tokio::time::sleep(Duration::from_millis(200)).await; + assert!( + !flush.is_finished(), + "an explicit flush must wait for the attempt lock rather than skip the attempt" + ); + attempt_lock + .commit() + .await + .expect("release the attempt lock"); + flush + .await + .expect("flush task") + .expect("flush progress after the lock is released"); + + let progress: Option = sqlx::query_scalar(&format!( + "SELECT progress FROM {schema}.attempt_state WHERE job_id = $1 AND run_lease = $2" + )) + .bind(job_id) + .bind(run_lease) + .fetch_optional(&pool) + .await + .expect("read attempt state"); + assert_eq!(progress, Some(serde_json::json!({"step": 3}))); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn test_queue_storage_receipt_rescue_skips_attempt_locked_for_completion() { let (_db_guard, pool) = setup_pool(10).await; From eeec5ea7427cc203474f25753aeed5a908626f2d Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Thu, 12 Nov 2026 08:06:23 +0000 Subject: [PATCH 2/4] docs(changelog): note the explicit progress flush fix --- CHANGELOG.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index de37a8d7a..ad43fac2f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,12 @@ Notable changes between releases. Detailed migration notes for storage transitions live in [`docs/upgrade-0.5-to-0.6.md`](docs/upgrade-0.5-to-0.6.md). +## [Unreleased] + +### Fixed + +- `JobContext::flush_progress` (Python `ctx.flush_progress()`) on a queue-storage attempt with receipts, the default, no longer returns success without writing the progress. A heartbeat leaves out an attempt whose lock another transaction holds, and since 0.6.16 an explicit flush shared that path, so a flush that overlapped a heartbeat or a completion of the same attempt was dropped and the worker treated the progress as persisted. An explicit flush now waits for the attempt's lock; heartbeats still skip. + ## [0.6.17] — 2026-11-12 Recommended for everyone on 0.6.16 using queue storage. Claims commit asynchronously and reuse the runtime's claimer lease, which raised throughput by 19 to 21% at 4, 16 and 64 workers and cut end-to-end p99 at 800 jobs/s from 31 ms to 24 ms in a three-round comparison against 0.6.16 on PostgreSQL 18 with `synchronous_commit = on`. Promotion and rescue no longer wait behind the leader's other maintenance work, a runtime can opt out of maintenance leadership, and several admin reads and the admin UI are fixed. See the [upgrade guide](docs/upgrade-0.6.16-to-0.6.17.md). From 317fe213929f13035a81922f8261cdaf57a656d7 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Thu, 12 Nov 2026 08:45:29 +0000 Subject: [PATCH 3/4] test(rescue): count recovery runs in the handler in the abandoned-backlog test The test waited for the retried jobs by counting completed rows in storage. Queue prune drops a sealed slot's completed rows once the ring wraps past it, so a recovery slower than one revolution could never reach the count. The handler now counts the recovery runs, and the test then waits for the queue to hold no running, retryable or available job. --- awa/tests/bulk_rescue_test.rs | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/awa/tests/bulk_rescue_test.rs b/awa/tests/bulk_rescue_test.rs index 8ce927567..89f41e5e5 100644 --- a/awa/tests/bulk_rescue_test.rs +++ b/awa/tests/bulk_rescue_test.rs @@ -801,13 +801,20 @@ async fn rescue_drains_an_abandoned_backlog_without_rescuing_live_claims() { .expect("recovery client") }; recovery.start().await.expect("start recovery client"); + // Completed rows leave the ring once queue prune wraps past them, so the + // handler counts the recovery runs. wait_until( scaled_timeout(Duration::from_secs(120)), - "the retried jobs to complete", + "every retried job to run", + || async { recovered.lock().expect("recovered").len() as i64 >= ABANDONED - exhausted }, + ) + .await; + wait_until( + scaled_timeout(Duration::from_secs(30)), + "the retried jobs to finish", || async { - count(&pool, format!( - "SELECT count(*) FROM awa.terminal_jobs WHERE queue = '{ABANDONED_QUEUE}' AND state = 'completed'" - )).await == ABANDONED - exhausted + running(&pool, &store, ABANDONED_QUEUE).await == 0 + && waiting_retry_counts(&pool, ABANDONED_QUEUE).await == (0, 0) }, ) .await; From b38b360c6c8c9c2a3d2e92a54c834c873f998c1c Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Thu, 12 Nov 2026 09:05:20 +0000 Subject: [PATCH 4/4] test(runtime-control): poll for the held session's observation until it exists Another test's stale-observation cleanup can delete the row between reports, so the poll reads it with fetch_optional instead of failing on RowNotFound. --- awa-worker/src/runtime_control.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/awa-worker/src/runtime_control.rs b/awa-worker/src/runtime_control.rs index 294fd06a3..3bc249af8 100644 --- a/awa-worker/src/runtime_control.rs +++ b/awa-worker/src/runtime_control.rs @@ -780,14 +780,14 @@ mod tests { assert!(!control.allowed()); tokio::time::timeout(Duration::from_secs(3), async { loop { - let value: Option = sqlx::query_scalar( + let value: Option> = sqlx::query_scalar( "SELECT observation FROM awa.runtime_control_observations WHERE instance_id=$1", ) .bind(id) - .fetch_one(&admin) + .fetch_optional(&admin) .await .unwrap(); - if value.is_some_and(|v| { + if value.flatten().is_some_and(|v| { v["observed_revision"] == halt.revision && v["state"] == "halted" }) { break;