From a41c85b59b411e3bfd78fb77a060017d2b8cfb55 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Sat, 7 Nov 2026 01:30:36 +1300 Subject: [PATCH 1/2] Fix moved-attempt hydration and protect point-read snapshots --- CHANGELOG.md | 1 + awa-model/src/admin.rs | 45 +++ awa-model/src/queue_storage.rs | 58 ++++ awa/tests/queue_storage_runtime_test.rs | 383 ++++++++++++++++++++++++ 4 files changed, 487 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 80fda9ffd..1eeedca4d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,7 @@ Notable changes between releases. Detailed migration notes for storage transitio ### Fixed +- Exclude tombstoned source rows when hydrating moved running jobs, including compact receipts, and preserve point-read snapshots against concurrent claim/lease partition pruning. - **Ring prune and terminal-count rollup report `Blocked` when any of their statements times out on a lock.** Prune sets a short `lock_timeout` (10 ms by default) before it takes `ACCESS EXCLUSIVE` on a slot's partitions, but diff --git a/awa-model/src/admin.rs b/awa-model/src/admin.rs index a40efb4e5..b45b2b8d3 100644 --- a/awa-model/src/admin.rs +++ b/awa-model/src/admin.rs @@ -381,6 +381,15 @@ pub fn queue_storage_current_jobs_cte(schema: &str) -> String { AND ready.run_lease + 1 = leases.run_lease AND ready.enqueue_shard = leases.enqueue_shard AND ready.lane_seq = leases.lane_seq + 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 + ) UNION ALL SELECT job_id, kind, queue, state, created_at, run_at, finalized_at FROM {schema}.terminal_jobs @@ -734,6 +743,15 @@ fn queue_storage_recent_jobs_cte(schema: &str, since: &str) -> String { AND ready.run_lease + 1 = leases.run_lease AND ready.enqueue_shard = leases.enqueue_shard AND ready.lane_seq = leases.lane_seq + 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 + ) WHERE ready.created_at >= {since} UNION ALL -- A queue or priority move copies the row with its times, so @@ -856,6 +874,15 @@ fn queue_storage_kind_counts_cte(schema: &str) -> String { AND ready.run_lease + 1 = leases.run_lease AND ready.enqueue_shard = leases.enqueue_shard AND ready.lane_seq = leases.lane_seq + 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 + ) GROUP BY ready.kind, leases.queue ), deferred_by_kind AS MATERIALIZED ( @@ -1090,6 +1117,15 @@ fn queue_storage_job_id_candidates_sql( AND ready.run_lease + 1 = leases.run_lease AND ready.enqueue_shard = leases.enqueue_shard AND ready.lane_seq = leases.lane_seq + 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 + ) WHERE TRUE{predicates} ORDER BY leases.job_id DESC LIMIT {limit} @@ -1552,6 +1588,15 @@ fn unique_key_candidate_sql(schema: &str, scoped: bool) -> String { AND ready.run_lease + 1 = leases.run_lease AND ready.enqueue_shard = leases.enqueue_shard AND ready.lane_seq = leases.lane_seq + 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 ready.unique_key = $1 OFFSET 0 ) AS ready ON TRUE diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index 0fb6aee36..b8c11dee5 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -11866,6 +11866,18 @@ impl QueueStorage { .await .map_err(map_sqlx_error)?; + // Protect running evidence from non-MVCC-safe TRUNCATE before + // the first data query establishes this repeatable-read snapshot. + // Claim-family order agrees with claim prune. + sqlx::query(&format!( + "LOCK TABLE {schema}.leases, {schema}.lease_claims, \ + {schema}.lease_claim_batches, {schema}.lease_claim_closures, \ + {schema}.lease_claim_closure_batches IN ACCESS SHARE MODE" + )) + .execute(tx.as_mut()) + .await + .map_err(map_sqlx_error)?; + let ready_rows: Vec = sqlx::query_as(&format!( r#" SELECT @@ -11979,6 +11991,15 @@ impl QueueStorage { AND ready.run_lease + 1 = lease.run_lease AND ready.enqueue_shard = lease.enqueue_shard AND ready.lane_seq = lease.lane_seq + 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 + ) LEFT JOIN {schema}.attempt_state AS attempt ON attempt.job_id = lease.job_id AND attempt.run_lease = lease.run_lease @@ -12039,6 +12060,15 @@ impl QueueStorage { AND ready.run_lease + 1 = claims.run_lease AND ready.enqueue_shard = claims.enqueue_shard AND ready.lane_seq = claims.lane_seq + 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 + ) LEFT JOIN {schema}.attempt_state AS attempt ON attempt.job_id = claims.job_id AND attempt.run_lease = claims.run_lease @@ -12115,6 +12145,16 @@ impl QueueStorage { AND ready.enqueue_shard = claim_batches.enqueue_shard AND ready.lane_seq = items.lane_seq AND ready.job_id = items.job_id + AND ready.run_lease + 1 = items.run_lease + 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 + ) LEFT JOIN {schema}.attempt_state AS attempt ON attempt.job_id = items.job_id AND attempt.run_lease = items.run_lease @@ -12567,6 +12607,15 @@ impl QueueStorage { AND ready.run_lease + 1 = lease.run_lease AND ready.enqueue_shard = lease.enqueue_shard AND ready.lane_seq = lease.lane_seq + 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 + ) LEFT JOIN {schema}.attempt_state AS attempt ON attempt.job_id = lease.job_id AND attempt.run_lease = lease.run_lease @@ -12766,6 +12815,15 @@ impl QueueStorage { AND ready.run_lease + 1 = lease.run_lease AND ready.enqueue_shard = lease.enqueue_shard AND ready.lane_seq = lease.lane_seq + 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 + ) LEFT JOIN {schema}.attempt_state AS attempt ON attempt.job_id = lease.job_id AND attempt.run_lease = lease.run_lease diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index e80c74c9e..681da1608 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -16848,3 +16848,386 @@ async fn test_point_read_uses_live_ready_queue_after_ring_wrap() { "point read must ignore the tombstoned source" ); } + +/// Aging moves a waiting job by tombstoning its ready row and re-inserting the +/// same attempt on the next lane. In a quiet queue the new lane hands out the +/// same `lane_seq`, so a lease joined to its ready row by attempt alone matched +/// both rows and the admin reads counted the running job twice. +async fn moved_ready_row_is_joined_once_by_attempt_reads(receipts: bool, row_local: bool) { + let (_db_guard, pool) = setup_pool(6).await; + let queue = "qs_moved_ready_lease_reads"; + let schema = if row_local { + "awa_qs_moved_ready_row_claim_reads" + } else if receipts { + "awa_qs_moved_ready_compact_reads" + } else { + "awa_qs_moved_ready_lease_reads" + }; + let store = create_store_with_config( + &pool, + QueueStorageConfig { + lease_claim_receipts: receipts, + ..unique_cancel_store_config(schema) + }, + ) + .await; + let job_id = enqueue_job( + &pool, + &store, + &unique_cancel_job("moved"), + InsertOpts { + priority: 2, + ..unique_cancel_opts(queue, None) + }, + ) + .await; + backdate_one_lane_ready_job_run_at( + &pool, + schema, + job_id, + Utc::now() - chrono::Duration::minutes(5), + ) + .await; + let aged = store + .age_waiting_priorities(&pool, Duration::from_secs(60), 10) + .await + .expect("age waiting priorities"); + assert_eq!(aged, vec![job_id]); + + let same_position: i64 = sqlx::query_scalar(&format!( + "SELECT count(DISTINCT priority)::bigint + FROM {schema}.ready_entries + WHERE job_id = $1 + GROUP BY ready_slot, ready_generation, enqueue_shard, lane_seq" + )) + .bind(job_id) + .fetch_one(&pool) + .await + .expect("ready rows for the moved job"); + assert_eq!( + same_position, 2, + "the source and destination rows must share a position for this test to mean anything" + ); + + let claimed = store + .claim_runtime_batch( + &pool, + queue, + 1, + if receipts && !row_local { + Duration::ZERO + } else { + Duration::from_secs(30) + }, + ) + .await + .expect("claim moved job"); + assert_eq!(claimed.len(), 1); + assert_eq!(claimed[0].job.id, job_id); + assert_eq!(claimed[0].claim.lease_claim_receipt, receipts); + assert_eq!( + claimed[0].claim.claim_batch_id.is_some(), + receipts && !row_local + ); + let legacy_rows: i64 = sqlx::query_scalar(&format!( + "SELECT count(*) FROM {schema}.lease_claims WHERE job_id = $1" + )) + .bind(job_id) + .fetch_one(&pool) + .await + .expect("verify the claim representation"); + assert_eq!(legacy_rows, i64::from(row_local)); + + // The release CTE projects materialized leases; receipt views use their + // existing open-claim sources. Do not add a new projection feature here. + if !receipts { + // This is the production CTE used by current-job projections, not only + // the point-read candidate list. The lease must hydrate exactly once. + let projected: Vec = sqlx::query_scalar(&format!( + "{} SELECT job_id FROM current_jobs WHERE queue = $1 AND state = 'running'", + admin::queue_storage_current_jobs_cte(schema) + )) + .bind(queue) + .fetch_all(&pool) + .await + .expect("project current running jobs"); + assert_eq!(projected, vec![job_id], "hydrate only the live ready row"); + } + + if !receipts { + let running = admin::list_jobs( + &pool, + &admin::ListJobsFilter { + state: Some(JobState::Running), + queue: Some(queue.to_string()), + ..Default::default() + }, + ) + .await + .expect("list running jobs"); + assert_eq!( + running.iter().map(|job| job.id).collect::>(), + vec![job_id], + "the moved running job is listed once" + ); + + let overview = admin::job_kind_overviews(&pool) + .await + .expect("job kind overviews"); + let kind = overview + .iter() + .find(|row| row.kind == UniqueCancelJob::kind()) + .expect("the moved job's kind is listed"); + assert_eq!(kind.job_count, 1, "the moved running job is counted once"); + } + + let loaded = store + .load_job(&pool, job_id) + .await + .expect("load moved job") + .expect("moved job exists"); + assert_eq!(loaded.state, JobState::Running); + assert_eq!(loaded.priority, 1, "reads the live destination row"); + assert_eq!( + loaded.metadata.get("_awa_original_priority"), + Some(&serde_json::json!(2)), + "payload comes from the aged destination, not its tombstoned source" + ); +} + +#[tokio::test] +async fn test_moved_ready_row_is_joined_once_by_lease_reads() { + moved_ready_row_is_joined_once_by_attempt_reads(false, false).await; +} + +#[tokio::test] +async fn test_moved_ready_row_is_joined_once_by_compact_receipt_reads() { + moved_ready_row_is_joined_once_by_attempt_reads(true, false).await; +} + +#[tokio::test] +async fn test_moved_ready_row_is_joined_once_by_row_local_receipt_reads() { + moved_ready_row_is_joined_once_by_attempt_reads(true, true).await; +} + +/// A point read's snapshot can outlive completion. TRUNCATE is not MVCC-safe: +/// pruning its claim/lease partition before the reader visits it would erase +/// the old running evidence, while the new completion is outside that snapshot. +async fn point_read_snapshot_survives_ring_prune(receipts: bool) { + let (_db_guard, pool) = setup_pool(6).await; + let schema = if receipts { + "awa_qs_snapshot_claim_prune" + } else { + "awa_qs_snapshot_lease_prune" + }; + let store = Arc::new( + create_store_with_config( + &pool, + QueueStorageConfig { + schema: schema.into(), + queue_slot_count: 4, + lease_slot_count: 2, + claim_slot_count: 2, + lease_claim_receipts: receipts, + ..Default::default() + }, + ) + .await, + ); + let queue = "snapshot_prune"; + let job_id = enqueue_job( + &pool, + &store, + &CompleteJob { id: 1 }, + InsertOpts { + queue: queue.into(), + ..Default::default() + }, + ) + .await; + let claimed = store + .claim_runtime_batch( + &pool, + queue, + 1, + if receipts { + Duration::ZERO + } else { + Duration::from_secs(30) + }, + ) + .await + .expect("claim the snapshot probe"); + assert_eq!(claimed.len(), 1); + assert_eq!(claimed[0].claim.claim_batch_id.is_some(), receipts); + assert_eq!( + store.load_job(&pool, job_id).await.unwrap().unwrap().state, + JobState::Running + ); + + // Gate the reader's second query without blocking completion's own + // deferred-state checks. Only this reader's application name takes the + // advisory lock; the view's scalar initplan runs even with no deferred rows. + sqlx::raw_sql(&format!( + r#"CREATE FUNCTION {schema}.snapshot_read_gate() RETURNS boolean + LANGUAGE plpgsql VOLATILE AS $$ + BEGIN + IF current_setting('application_name') = 'awa_snapshot_reader' THEN + PERFORM pg_advisory_xact_lock(60006, 1); + END IF; + RETURN TRUE; + END; + $$; + ALTER TABLE {schema}.deferred_jobs RENAME TO deferred_rows; + CREATE VIEW {schema}.deferred_jobs AS + SELECT * FROM {schema}.deferred_rows + WHERE (SELECT {schema}.snapshot_read_gate())"# + )) + .execute(&pool) + .await + .expect("install reader-only query gate"); + let mut gate = pool.begin().await.expect("begin read gate"); + sqlx::query("SELECT pg_advisory_xact_lock(60006, 1)") + .execute(gate.as_mut()) + .await + .expect("hold the second read"); + let reader_pool = PgPoolOptions::new() + .max_connections(1) + .connect_with( + (*pool.connect_options()) + .clone() + .application_name("awa_snapshot_reader"), + ) + .await + .expect("connect isolated snapshot reader"); + let reader_store = store.clone(); + let reader_connection = reader_pool.clone(); + let reader = + tokio::spawn(async move { reader_store.load_job(&reader_connection, job_id).await }); + let deadline = Instant::now() + ci_timing::scaled_timeout(Duration::from_secs(10)); + loop { + let at_gate: bool = sqlx::query_scalar( + "SELECT EXISTS ( + SELECT 1 FROM pg_locks + WHERE locktype = 'advisory' + AND database = (SELECT oid FROM pg_database WHERE datname = current_database()) + AND classid = 60006 AND objid = 1 AND NOT granted + )", + ) + .fetch_one(&pool) + .await + .expect("observe the reader-only gate"); + if at_gate { + break; + } + assert!( + Instant::now() < deadline, + "point read did not reach its second query" + ); + tokio::time::sleep(Duration::from_millis(10)).await; + } + + let during_read = + tokio::time::timeout(ci_timing::scaled_timeout(Duration::from_secs(10)), async { + assert_eq!( + store + .complete_runtime_batch(&pool, &claimed) + .await + .expect("complete outside the snapshot") + .len(), + 1 + ); + let rotation = if receipts { + store.rotate_claims(&pool).await + } else { + store.rotate_leases(&pool).await + } + .expect("seal the completed attempt's slot"); + assert!(matches!(rotation, RotateOutcome::Rotated { slot: 1, .. })); + if receipts { + store + .prune_oldest_claims(&pool) + .await + .expect("prune while the snapshot is live") + } else { + // The release lease-pruner skips an already empty child. Use + // the actual destructive SQL here to exercise its MVCC hazard. + let mut prune = pool.begin().await.expect("begin lease truncate"); + sqlx::query("SET LOCAL lock_timeout = '10ms'") + .execute(prune.as_mut()) + .await + .expect("bound lease truncate"); + let result = sqlx::query(&format!("TRUNCATE TABLE {schema}.leases_0")) + .execute(prune.as_mut()) + .await; + match result { + Ok(_) => { + prune.commit().await.expect("commit lease truncate"); + PruneOutcome::Pruned { + slot: 0, + carried_failed_rows: 0, + } + } + Err(sqlx::Error::Database(error)) + if error.code().as_deref() == Some("55P03") => + { + prune + .rollback() + .await + .expect("release blocked lease truncate"); + PruneOutcome::Blocked { slot: 0 } + } + Err(error) => panic!("lease truncate failed: {error}"), + } + } + }) + .await; + + // Always release the gate and finish the reader before asserting the + // outcome, so the negative control also reports the inconsistent state. + gate.rollback().await.expect("release the read gate"); + let snapshot = tokio::time::timeout(ci_timing::scaled_timeout(Duration::from_secs(10)), reader) + .await + .expect("point read finished") + .expect("join point read") + .expect("point read succeeds") + .expect("the job remains in its snapshot"); + let during_read = + during_read.expect("completion/prune must not wait on the deferred read gate"); + assert_eq!( + ( + matches!(during_read, PruneOutcome::Blocked { slot: 0 }), + snapshot.state + ), + (true, JobState::Running), + "prune must retain the snapshot's running evidence: {during_read:?}" + ); + assert_eq!(snapshot.run_lease, claimed[0].job.run_lease); + reader_pool.close().await; + if receipts { + let after_read = store + .prune_oldest_claims(&pool) + .await + .expect("prune after the reader releases its snapshot"); + assert!(matches!(after_read, PruneOutcome::Pruned { slot: 0, .. })); + } else { + sqlx::query(&format!("TRUNCATE TABLE {schema}.leases_0")) + .execute(&pool) + .await + .expect("truncate after the reader releases its snapshot"); + } + assert_eq!( + store.load_job(&pool, job_id).await.unwrap().unwrap().state, + JobState::Completed + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn test_point_read_snapshot_blocks_compact_claim_prune() { + point_read_snapshot_survives_ring_prune(true).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn test_point_read_snapshot_blocks_lease_truncate() { + point_read_snapshot_survives_ring_prune(false).await; +} From 9b05fe2b79dece8cc5a6d746ff847e865001ebe5 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Sat, 7 Nov 2026 09:30:17 +1300 Subject: [PATCH 2/2] fix: lock ready ring before point-read snapshots --- CHANGELOG.md | 2 + awa-model/src/queue_storage.rs | 10 +- awa/tests/queue_storage_runtime_test.rs | 209 ++++++++++++++++++++++++ 3 files changed, 217 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1eeedca4d..7a0f42542 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,8 @@ Notable changes between releases. Detailed migration notes for storage transitio ### Fixed +- Lock the ready ring before point reads establish their snapshot, so + concurrent queue pruning cannot hide a failed job carried to a live slot. - Exclude tombstoned source rows when hydrating moved running jobs, including compact receipts, and preserve point-read snapshots against concurrent claim/lease partition pruning. - **Ring prune and terminal-count rollup report `Blocked` when any of their statements times out on a lock.** Prune sets a short `lock_timeout` (10 ms diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index b8c11dee5..15428b227 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -11866,11 +11866,13 @@ impl QueueStorage { .await .map_err(map_sqlx_error)?; - // Protect running evidence from non-MVCC-safe TRUNCATE before - // the first data query establishes this repeatable-read snapshot. - // Claim-family order agrees with claim prune. + // Protect all rings prune can truncate before the first data query + // establishes this repeatable-read snapshot: the query can take its + // snapshot before acquiring its relation locks. Queue prune takes the + // ready child NOWAIT before terminal children, so ready also fences + // those. Claim-family order agrees with claim prune. sqlx::query(&format!( - "LOCK TABLE {schema}.leases, {schema}.lease_claims, \ + "LOCK TABLE {schema}.ready_entries, {schema}.leases, {schema}.lease_claims, \ {schema}.lease_claim_batches, {schema}.lease_claim_closures, \ {schema}.lease_claim_closure_batches IN ACCESS SHARE MODE" )) diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index 681da1608..d3751bd74 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -17231,3 +17231,212 @@ async fn test_point_read_snapshot_blocks_compact_claim_prune() { async fn test_point_read_snapshot_blocks_lease_truncate() { point_read_snapshot_survives_ring_prune(false).await; } + +/// A reader's first query can establish its snapshot before waiting on its +/// ready relation lock. Queue prune must finish its failed-row carry before +/// that snapshot, or TRUNCATE hides the source and the new carried version. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_point_read_first_snapshot_follows_ready_prune_carry() { + let (_db_guard, pool) = setup_pool(10).await; + let schema = "awa_qs_first_snapshot_ready_carry"; + let queue = "first_snapshot_ready_carry"; + let bound = ci_timing::scaled_timeout(Duration::from_secs(10)); + let store = Arc::new( + create_store(&pool, schema) + .await + .with_prune_lock_timeout(ci_timing::scaled_timeout(Duration::from_secs(20))) + .expect("bound the controlled prune lock wait"), + ); + let job_id = enqueue_job( + &pool, + &store, + &CompleteJob { id: 315 }, + InsertOpts { + queue: queue.into(), + metadata: serde_json::json!({"ready_prune_guard": "retained payload"}), + ..Default::default() + }, + ) + .await; + let claimed = store + .claim_runtime_batch(&pool, queue, 1, Duration::from_secs(300)) + .await + .expect("claim the retained failed job"); + assert_eq!(claimed.len(), 1); + let run_lease = claimed[0].job.run_lease; + store + .fail_terminal(&pool, job_id, run_lease, "retained failure", None) + .await + .expect("fail the retained job") + .expect("the attempt was running"); + let before = store + .load_job(&pool, job_id) + .await + .expect("load the original failed job") + .expect("the failed job exists"); + assert_eq!(before.state, JobState::Failed); + assert!(matches!( + store.rotate(&pool).await.expect("seal slot zero"), + RotateOutcome::Rotated { slot: 1, .. } + )); + + // Distinct one-connection pools keep the observed backend identities + // stable across the two public operations and their relation waits. + let pruner_pool = PgPoolOptions::new() + .max_connections(1) + .connect_with( + (*pool.connect_options()) + .clone() + .application_name("awa_ready_carry_pruner"), + ) + .await + .expect("connect the controlled pruner"); + let reader_pool = PgPoolOptions::new() + .max_connections(1) + .connect_with( + (*pool.connect_options()) + .clone() + .application_name("awa_ready_carry_reader"), + ) + .await + .expect("connect the controlled reader"); + let pruner_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&pruner_pool) + .await + .expect("identify the pruner"); + let reader_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&reader_pool) + .await + .expect("identify the reader"); + + let mut carry_gate = pool.begin().await.expect("begin the carry gate"); + sqlx::query(&format!("LOCK TABLE {schema}.done_entries_1 IN SHARE MODE")) + .execute(carry_gate.as_mut()) + .await + .expect("hold the live carry destination"); + let prune_store = store.clone(); + let prune_connection = pruner_pool.clone(); + let mut pruner = tokio::spawn(async move { + prune_store + .prune_oldest(&prune_connection, Duration::from_secs(3600)) + .await + }); + let mut reader = None; + let wait_sql = r#" + SELECT activity.query + FROM pg_stat_activity AS activity + JOIN pg_locks AS locks ON locks.pid = activity.pid + WHERE activity.pid = $1 + AND activity.datid = (SELECT oid FROM pg_database + WHERE datname = current_database()) + AND activity.state = 'active' + AND activity.wait_event_type = 'Lock' + AND activity.query LIKE $3 + AND locks.locktype = 'relation' + AND locks.relation = to_regclass($2) + AND NOT locks.granted + LIMIT 1 + "#; + let observed = tokio::time::timeout(bound, async { + loop { + let waiting: Option = sqlx::query_scalar(wait_sql) + .bind(pruner_pid) + .bind(format!("{schema}.done_entries_1")) + .bind(format!("%INSERT INTO {schema}.done_entries%")) + .fetch_optional(&pool) + .await?; + if waiting.is_some() { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + + let read_store = store.clone(); + let read_connection = reader_pool.clone(); + reader = Some(tokio::spawn(async move { + read_store.load_job(&read_connection, job_id).await + })); + loop { + let waiting: Option = sqlx::query_scalar(wait_sql) + .bind(reader_pid) + .bind(format!("{schema}.ready_entries_0")) + .bind(format!("%{schema}.ready_entries%")) + .fetch_optional(&pool) + .await?; + if let Some(query) = waiting { + return Ok::<_, sqlx::Error>(query); + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await; + + // Release every gate and join both tasks before outcome assertions, so + // the old-source negative control cannot strand a read or lock holder. + let released = tokio::time::timeout(bound, carry_gate.rollback()).await; + let prune_result = match tokio::time::timeout(bound, &mut pruner).await { + Ok(result) => Some(result), + Err(_) => { + pruner.abort(); + let _ = pruner.await; + None + } + }; + let read_result = match reader { + Some(mut task) => match tokio::time::timeout(bound, &mut task).await { + Ok(result) => Some(result), + Err(_) => { + task.abort(); + let _ = task.await; + None + } + }, + None => None, + }; + let pruner_closed = tokio::time::timeout(bound, pruner_pool.close()).await; + let reader_closed = tokio::time::timeout(bound, reader_pool.close()).await; + + released + .expect("release the carry gate within the bound") + .expect("roll back the carry gate"); + pruner_closed.expect("close the pruner pool"); + reader_closed.expect("close the reader pool"); + let waiting_query = observed + .expect("observe both exact relation waits within the bound") + .expect("query the relation waits"); + let pruned = prune_result + .expect("prune finishes within the bound") + .expect("join the pruner") + .expect("prune succeeds after the carry gate releases"); + assert!( + matches!( + pruned, + PruneOutcome::Pruned { + slot: 0, + carried_failed_rows: 1, + .. + } + ), + "the real prune must carry the failed row: {pruned:?}" + ); + let loaded = read_result + .expect("reader finishes within the bound") + .expect("join the reader") + .expect("the public read succeeds") + .unwrap_or_else(|| { + panic!("retained failed job disappeared; reader waited in {waiting_query}") + }); + let fresh = store + .load_job(&pool, job_id) + .await + .expect("load after prune commits") + .expect("the retained failed job still exists"); + for row in [&loaded, &fresh] { + assert_eq!(row.id, job_id); + assert_eq!(row.state, JobState::Failed); + assert_eq!(row.run_lease, run_lease); + assert_eq!(row.args, before.args); + assert_eq!(row.metadata, before.metadata); + assert_eq!(row.finalized_at, before.finalized_at); + } +}