diff --git a/CHANGELOG.md b/CHANGELOG.md index 54759f505..820d8d60e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,17 @@ Notable changes between releases. Detailed migration notes for storage transitio ## [Unreleased] +### Fixed + +- `awa.jobs` keeps running jobs visible after priority aging changes their + effective priority. Migration v053 joins receipt claims and materialized leases + to the exact job attempt rather than comparing effective and lane priorities, + excluding tombstoned and older-attempt ready rows to prevent duplicates. + Receipt and materialized-lease rows consistently report the claimed effective + priority. + The migration only refreshes the compatibility function; storage and runtime + write protocols are unchanged. + ## [0.7.0-alpha.1] — 2026-10-29 ### Upgrading diff --git a/awa-model/migrations/v053_jobs_compat_priority_aging.sql b/awa-model/migrations/v053_jobs_compat_priority_aging.sql new file mode 100644 index 000000000..3ecb405d7 --- /dev/null +++ b/awa-model/migrations/v053_jobs_compat_priority_aging.sql @@ -0,0 +1,564 @@ +-- v053: keep aged running jobs visible through awa.jobs. +-- Issue: https://github.com/hardbyte/awa/issues/569. +-- +-- Claims and leases store effective priority after aging; ready entries keep +-- their original lane priority. Match the job and claimed attempt instead of +-- comparing those two priority domains. Exclude tombstoned ready entries so +-- reprioritization cannot duplicate a running row, and match run_lease so a +-- retained ready entry from an earlier attempt cannot supply its body. +-- +-- N-1 compatibility: only refreshes awa.jobs_compat(), preserving its return +-- type and all storage/write protocols. Existing binaries keep using the same +-- view and claim functions; no new runtime capability or preflight is needed. +-- +-- Lock impact: brief function-catalog and schema-version writes only, with no +-- hot-table DDL or data scan. Execution time is independent of queue depth. +-- Re-runnable and transaction-safe. External runners apply this bundle normally. + +CREATE OR REPLACE FUNCTION awa.jobs_compat() +RETURNS TABLE ( + id BIGINT, + kind TEXT, + queue TEXT, + args JSONB, + state awa.job_state, + priority SMALLINT, + attempt SMALLINT, + max_attempts SMALLINT, + run_at TIMESTAMPTZ, + heartbeat_at TIMESTAMPTZ, + deadline_at TIMESTAMPTZ, + attempted_at TIMESTAMPTZ, + finalized_at TIMESTAMPTZ, + created_at TIMESTAMPTZ, + errors JSONB[], + metadata JSONB, + tags TEXT[], + unique_key BYTEA, + unique_states BIT(8), + callback_id UUID, + callback_timeout_at TIMESTAMPTZ, + callback_filter TEXT, + callback_on_complete TEXT, + callback_on_fail TEXT, + callback_transform TEXT, + run_lease BIGINT, + progress JSONB +) +LANGUAGE plpgsql +STABLE +SET search_path = pg_catalog, awa, public +AS $$ +DECLARE + v_schema TEXT; +BEGIN + v_schema := awa.active_queue_storage_schema(); + + IF v_schema IS NULL THEN + RETURN QUERY + SELECT + j.id, + j.kind, + j.queue, + j.args, + j.state, + j.priority, + j.attempt, + j.max_attempts, + j.run_at, + j.heartbeat_at, + j.deadline_at, + j.attempted_at, + j.finalized_at, + j.created_at, + j.errors, + j.metadata, + j.tags, + j.unique_key, + j.unique_states, + j.callback_id, + j.callback_timeout_at, + j.callback_filter, + j.callback_on_complete, + j.callback_on_fail, + j.callback_transform, + j.run_lease, + j.progress + FROM awa.jobs_hot AS j + UNION ALL + SELECT + j.id, + j.kind, + j.queue, + j.args, + j.state, + j.priority, + j.attempt, + j.max_attempts, + j.run_at, + j.heartbeat_at, + j.deadline_at, + j.attempted_at, + j.finalized_at, + j.created_at, + j.errors, + j.metadata, + j.tags, + j.unique_key, + j.unique_states, + j.callback_id, + j.callback_timeout_at, + j.callback_filter, + j.callback_on_complete, + j.callback_on_fail, + j.callback_transform, + j.run_lease, + j.progress + FROM awa.scheduled_jobs AS j; + RETURN; + END IF; + + RETURN QUERY EXECUTE format( + $sql$ + WITH claim_cursors AS MATERIALIZED ( + -- One catalog read per lane instead of one per ready row + -- (#422). sequence_next_value is VOLATILE, so spelled inline in + -- the available-branch filter the planner could neither use it + -- as an index bound nor cache it across rows: every + -- ready_entries child was seq-scanned in full and each surviving + -- row paid a dynamic-SQL catalog round-trip (~15x slower on a + -- 20k-row backlog). + SELECT + claims.queue, + claims.priority, + claims.enqueue_shard, + %1$I.sequence_next_value(claims.seq_name) AS claim_seq + FROM %1$I.queue_claim_heads AS claims + ), + current_available AS ( + SELECT + ready.job_id AS id, + ready.kind, + ready.queue, + ready.args, + 'available'::awa.job_state AS state, + ready.priority, + ready.attempt, + ready.max_attempts, + ready.run_at, + NULL::timestamptz AS heartbeat_at, + NULL::timestamptz AS deadline_at, + ready.attempted_at, + NULL::timestamptz AS finalized_at, + ready.created_at, + awa.queue_storage_payload_errors(ready.payload) AS errors, + COALESCE(NULLIF(ready.payload->'metadata', 'null'::jsonb), '{}'::jsonb) AS metadata, + awa.queue_storage_payload_tags(ready.payload) AS tags, + ready.unique_key, + CASE + WHEN ready.unique_states IS NULL THEN NULL::bit(8) + ELSE ready.unique_states::bit(8) + END AS unique_states, + NULL::uuid AS callback_id, + NULL::timestamptz AS callback_timeout_at, + NULL::text AS callback_filter, + NULL::text AS callback_on_complete, + NULL::text AS callback_on_fail, + NULL::text AS callback_transform, + ready.run_lease, + NULLIF(ready.payload->'progress', 'null'::jsonb) AS progress + FROM %1$I.ready_entries AS ready + JOIN claim_cursors AS claims + ON claims.queue = ready.queue + AND claims.priority = ready.priority + AND claims.enqueue_shard = ready.enqueue_shard + WHERE ready.lane_seq >= claims.claim_seq + AND NOT EXISTS ( + SELECT 1 + FROM %1$I.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 + ) + ) + SELECT + current_available.id, + current_available.kind, + current_available.queue, + current_available.args, + current_available.state, + current_available.priority, + current_available.attempt, + current_available.max_attempts, + current_available.run_at, + current_available.heartbeat_at, + current_available.deadline_at, + current_available.attempted_at, + current_available.finalized_at, + current_available.created_at, + current_available.errors, + current_available.metadata, + current_available.tags, + current_available.unique_key, + current_available.unique_states, + current_available.callback_id, + current_available.callback_timeout_at, + current_available.callback_filter, + current_available.callback_on_complete, + current_available.callback_on_fail, + current_available.callback_transform, + current_available.run_lease, + current_available.progress + FROM current_available + UNION ALL + -- Receipt claims that have not materialised into `leases` are + -- running too (#246 / #416 / #422): legacy row-local `lease_claims` + -- below, default compact `lease_claim_batches` members next. Each + -- leg anti-joins every durable closure/supersession shape so a job + -- reports at most one state. Mirrors admin::state_counts (#410) and + -- queue_storage::open_receipt_running_claims_sql. + SELECT + claims.job_id AS id, + ready.kind, + ready.queue, + ready.args, + 'running'::awa.job_state AS state, + claims.priority, + claims.attempt, + claims.max_attempts, + ready.run_at, + attempt.heartbeat_at, + claims.deadline_at, + claims.claimed_at AS attempted_at, + NULL::timestamptz AS finalized_at, + ready.created_at, + awa.queue_storage_payload_errors(ready.payload) AS errors, + COALESCE(NULLIF(ready.payload->'metadata', 'null'::jsonb), '{}'::jsonb) AS metadata, + awa.queue_storage_payload_tags(ready.payload) AS tags, + ready.unique_key, + CASE + WHEN ready.unique_states IS NULL THEN NULL::bit(8) + ELSE ready.unique_states::bit(8) + END AS unique_states, + NULL::uuid AS callback_id, + NULL::timestamptz AS callback_timeout_at, + NULL::text AS callback_filter, + NULL::text AS callback_on_complete, + NULL::text AS callback_on_fail, + NULL::text AS callback_transform, + claims.run_lease, + COALESCE(NULLIF(attempt.progress, 'null'::jsonb), NULLIF(ready.payload->'progress', 'null'::jsonb)) + FROM %1$I.lease_claims AS claims + JOIN %1$I.ready_entries AS ready + ON ready.ready_slot = claims.ready_slot + AND ready.ready_generation = claims.ready_generation + AND ready.queue = claims.queue + AND ready.run_lease + 1 = claims.run_lease + AND ready.enqueue_shard = claims.enqueue_shard + AND ready.lane_seq = claims.lane_seq + AND ready.job_id = claims.job_id + AND NOT EXISTS ( + SELECT 1 FROM %1$I.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 %1$I.attempt_state AS attempt + ON attempt.job_id = claims.job_id + AND attempt.run_lease = claims.run_lease + WHERE claims.closed_at IS NULL + AND NOT EXISTS ( + SELECT 1 FROM %1$I.lease_claim_closures AS cx + WHERE cx.claim_slot = claims.claim_slot + AND cx.job_id = claims.job_id + AND cx.run_lease = claims.run_lease + ) + AND NOT EXISTS ( + SELECT 1 FROM %1$I.lease_claim_closure_batches AS cb + WHERE cb.receipt_ranges @> claims.receipt_id + ) + AND NOT EXISTS ( + SELECT 1 FROM %1$I.leases AS lease + WHERE lease.job_id = claims.job_id + AND lease.run_lease = claims.run_lease + ) + AND NOT EXISTS ( + SELECT 1 FROM %1$I.done_entries AS done + WHERE done.job_id = claims.job_id + AND done.run_lease = claims.run_lease + ) + AND NOT EXISTS ( + SELECT 1 FROM %1$I.deferred_jobs AS deferred + WHERE deferred.job_id = claims.job_id + AND deferred.run_lease = claims.run_lease + ) + AND NOT EXISTS ( + SELECT 1 FROM %1$I.dlq_entries AS dlq + WHERE dlq.job_id = claims.job_id + AND dlq.run_lease = claims.run_lease + ) + UNION ALL + SELECT + items.job_id AS id, + ready.kind, + ready.queue, + ready.args, + 'running'::awa.job_state AS state, + batches.priority, + batches.attempts[items.ord], + batches.max_attempts[items.ord], + ready.run_at, + attempt.heartbeat_at, + batches.deadline_at, + batches.claimed_at AS attempted_at, + NULL::timestamptz AS finalized_at, + ready.created_at, + awa.queue_storage_payload_errors(ready.payload) AS errors, + COALESCE(NULLIF(ready.payload->'metadata', 'null'::jsonb), '{}'::jsonb) AS metadata, + awa.queue_storage_payload_tags(ready.payload) AS tags, + ready.unique_key, + CASE + WHEN ready.unique_states IS NULL THEN NULL::bit(8) + ELSE ready.unique_states::bit(8) + END AS unique_states, + NULL::uuid AS callback_id, + NULL::timestamptz AS callback_timeout_at, + NULL::text AS callback_filter, + NULL::text AS callback_on_complete, + NULL::text AS callback_on_fail, + NULL::text AS callback_transform, + items.run_lease, + COALESCE(NULLIF(attempt.progress, 'null'::jsonb), NULLIF(ready.payload->'progress', 'null'::jsonb)) + FROM %1$I.lease_claim_batches AS batches + CROSS JOIN LATERAL unnest( + batches.job_ids, + batches.run_leases, + batches.receipt_ids, + batches.lane_seqs + ) WITH ORDINALITY AS items(job_id, run_lease, receipt_id, lane_seq, ord) + JOIN %1$I.ready_entries AS ready + ON ready.ready_slot = batches.ready_slot + AND ready.ready_generation = batches.ready_generation + AND ready.queue = batches.queue + AND ready.run_lease + 1 = items.run_lease + AND ready.enqueue_shard = batches.enqueue_shard + AND ready.lane_seq = items.lane_seq + AND ready.job_id = items.job_id + AND NOT EXISTS ( + SELECT 1 FROM %1$I.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 %1$I.attempt_state AS attempt + ON attempt.job_id = items.job_id + AND attempt.run_lease = items.run_lease + WHERE NOT EXISTS ( + SELECT 1 FROM %1$I.lease_claim_closures AS cx + WHERE cx.claim_slot = batches.claim_slot + AND cx.job_id = items.job_id + AND cx.run_lease = items.run_lease + ) + AND NOT EXISTS ( + SELECT 1 FROM %1$I.lease_claim_closure_batches AS cb + WHERE cb.claim_slot = batches.claim_slot + AND cb.receipt_ranges @> items.receipt_id + ) + AND NOT EXISTS ( + SELECT 1 FROM %1$I.leases AS lease + WHERE lease.job_id = items.job_id + AND lease.run_lease = items.run_lease + ) + AND NOT EXISTS ( + SELECT 1 FROM %1$I.done_entries AS done + WHERE done.job_id = items.job_id + AND done.run_lease = items.run_lease + ) + AND NOT EXISTS ( + SELECT 1 FROM %1$I.deferred_jobs AS deferred + WHERE deferred.job_id = items.job_id + AND deferred.run_lease = items.run_lease + ) + AND NOT EXISTS ( + SELECT 1 FROM %1$I.dlq_entries AS dlq + WHERE dlq.job_id = items.job_id + AND dlq.run_lease = items.run_lease + ) + UNION ALL + SELECT + deferred.job_id AS id, + deferred.kind, + deferred.queue, + deferred.args, + deferred.state, + deferred.priority, + deferred.attempt, + deferred.max_attempts, + deferred.run_at, + NULL::timestamptz AS heartbeat_at, + NULL::timestamptz AS deadline_at, + deferred.attempted_at, + deferred.finalized_at, + deferred.created_at, + awa.queue_storage_payload_errors(deferred.payload) AS errors, + COALESCE(NULLIF(deferred.payload->'metadata', 'null'::jsonb), '{}'::jsonb) AS metadata, + awa.queue_storage_payload_tags(deferred.payload) AS tags, + deferred.unique_key, + CASE + WHEN deferred.unique_states IS NULL THEN NULL::bit(8) + ELSE deferred.unique_states::bit(8) + END AS unique_states, + NULL::uuid AS callback_id, + NULL::timestamptz AS callback_timeout_at, + NULL::text AS callback_filter, + NULL::text AS callback_on_complete, + NULL::text AS callback_on_fail, + NULL::text AS callback_transform, + deferred.run_lease, + NULLIF(deferred.payload->'progress', 'null'::jsonb) AS progress + FROM %1$I.deferred_jobs AS deferred + UNION ALL + SELECT + leases.job_id AS id, + ready.kind, + ready.queue, + ready.args, + leases.state, + leases.priority, + leases.attempt, + leases.max_attempts, + ready.run_at, + leases.heartbeat_at, + leases.deadline_at, + leases.attempted_at, + NULL::timestamptz AS finalized_at, + ready.created_at, + awa.queue_storage_payload_errors(ready.payload) AS errors, + CASE + WHEN attempt.callback_result IS NULL + THEN COALESCE(NULLIF(ready.payload->'metadata', 'null'::jsonb), '{}'::jsonb) + ELSE COALESCE(NULLIF(ready.payload->'metadata', 'null'::jsonb), '{}'::jsonb) + || jsonb_build_object('_awa_callback_result', attempt.callback_result) + END AS metadata, + awa.queue_storage_payload_tags(ready.payload) AS tags, + ready.unique_key, + CASE + WHEN ready.unique_states IS NULL THEN NULL::bit(8) + ELSE ready.unique_states::bit(8) + END AS unique_states, + leases.callback_id, + leases.callback_timeout_at, + attempt.callback_filter, + attempt.callback_on_complete, + attempt.callback_on_fail, + attempt.callback_transform, + leases.run_lease, + COALESCE( + NULLIF(attempt.progress, 'null'::jsonb), + NULLIF(ready.payload->'progress', 'null'::jsonb) + ) AS progress + FROM %1$I.leases AS leases + JOIN %1$I.ready_entries AS ready + ON ready.ready_slot = leases.ready_slot + AND ready.ready_generation = leases.ready_generation + AND ready.queue = leases.queue + AND ready.run_lease + 1 = leases.run_lease + AND ready.enqueue_shard = leases.enqueue_shard + AND ready.lane_seq = leases.lane_seq + AND ready.job_id = leases.job_id + AND NOT EXISTS ( + SELECT 1 FROM %1$I.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 %1$I.attempt_state AS attempt + ON attempt.job_id = leases.job_id + AND attempt.run_lease = leases.run_lease + UNION ALL + SELECT + done.job_id AS id, + done.kind, + done.queue, + done.args, + done.state, + done.priority, + done.attempt, + done.max_attempts, + done.run_at, + NULL::timestamptz AS heartbeat_at, + NULL::timestamptz AS deadline_at, + done.attempted_at, + done.finalized_at, + done.created_at, + awa.queue_storage_payload_errors(done.payload) AS errors, + COALESCE(NULLIF(done.payload->'metadata', 'null'::jsonb), '{}'::jsonb) AS metadata, + awa.queue_storage_payload_tags(done.payload) AS tags, + done.unique_key, + CASE + WHEN done.unique_states IS NULL THEN NULL::bit(8) + ELSE done.unique_states::bit(8) + END AS unique_states, + NULL::uuid AS callback_id, + NULL::timestamptz AS callback_timeout_at, + NULL::text AS callback_filter, + NULL::text AS callback_on_complete, + NULL::text AS callback_on_fail, + NULL::text AS callback_transform, + done.run_lease, + NULLIF(done.payload->'progress', 'null'::jsonb) AS progress + FROM %1$I.terminal_jobs AS done + UNION ALL + SELECT + dlq.job_id AS id, + dlq.kind, + dlq.queue, + dlq.args, + dlq.state, + dlq.priority, + dlq.attempt, + dlq.max_attempts, + dlq.run_at, + NULL::timestamptz AS heartbeat_at, + NULL::timestamptz AS deadline_at, + dlq.attempted_at, + dlq.finalized_at, + dlq.created_at, + awa.queue_storage_payload_errors(dlq.payload) AS errors, + COALESCE(NULLIF(dlq.payload->'metadata', 'null'::jsonb), '{}'::jsonb) AS metadata, + awa.queue_storage_payload_tags(dlq.payload) AS tags, + dlq.unique_key, + CASE + WHEN dlq.unique_states IS NULL THEN NULL::bit(8) + ELSE dlq.unique_states::bit(8) + END AS unique_states, + NULL::uuid AS callback_id, + NULL::timestamptz AS callback_timeout_at, + NULL::text AS callback_filter, + NULL::text AS callback_on_complete, + NULL::text AS callback_on_fail, + NULL::text AS callback_transform, + dlq.run_lease, + NULLIF(dlq.payload->'progress', 'null'::jsonb) AS progress + FROM %1$I.dlq_entries AS dlq + $sql$, + v_schema + ); +END; +$$; + +INSERT INTO awa.schema_version (version, description) +VALUES (53, 'Keep aged running jobs visible in jobs_compat()') +ON CONFLICT (version) DO NOTHING; diff --git a/awa-model/src/migrations.rs b/awa-model/src/migrations.rs index 385035b65..fff4e4222 100644 --- a/awa-model/src/migrations.rs +++ b/awa-model/src/migrations.rs @@ -5,7 +5,7 @@ use sqlx::{Connection, PgPool}; use tracing::{info, warn}; /// Current schema version. -pub const CURRENT_VERSION: i32 = 52; +pub const CURRENT_VERSION: i32 = 53; /// Migrations that require an exclusive (no-live-runtime) upgrade window. /// @@ -292,6 +292,11 @@ const MIGRATIONS: &[(i32, &str, &[&str])] = &[ ), (51, "Per-slot claim functions", &[V23_UP, V51_UP]), (52, "Retained job output", &[V23_UP, V52_UP]), + ( + 53, + "Keep aged running jobs visible in jobs_compat()", + &[V53_UP], + ), ]; const V1_UP: &str = include_str!("../migrations/v001_canonical_schema.sql"); @@ -352,6 +357,7 @@ const V49_UP: &str = include_str!("../migrations/v049_prepared_lane_sequences.sq const V50_UP: &str = include_str!("../migrations/v050_drop_ready_lane_shard_index.sql"); const V51_UP: &str = include_str!("../migrations/v051_claim_slot_functions.sql"); const V52_UP: &str = include_str!("../migrations/v052_retained_job_output.sql"); +const V53_UP: &str = include_str!("../migrations/v053_jobs_compat_priority_aging.sql"); /// Old version numbers from pre-0.4 releases that used V3/V4/V5 numbering. /// Also tolerates the unreleased inline-V6 branch numbering used during review. diff --git a/awa/tests/migration_test.rs b/awa/tests/migration_test.rs index 53c1ab009..db5d91d24 100644 --- a/awa/tests/migration_test.rs +++ b/awa/tests/migration_test.rs @@ -381,6 +381,131 @@ async fn test_fresh_install_reaches_current_version() { ); } +#[tokio::test] +async fn test_v053_preserves_aged_running_claims_during_upgrade() { + let _guard = acquire_migration_guard().await; + let pool = pool().await; + reset_schema(&pool).await; + for (_, _, sql) in migrations::migration_sql_range(0, 52) { + sqlx::raw_sql(audited_sql(sql)) + .execute(&pool) + .await + .expect("install the actual v052 schema"); + } + simulate_non_canonical_compat_routing(&pool).await; + assert_eq!(migrations::current_version(&pool).await.unwrap(), 52); + let store = QueueStorage::new(awa::model::QueueStorageConfig { + schema: "awa_queue_storage".to_string(), + lease_claim_receipts: true, + ..Default::default() + }) + .unwrap(); + let queue = "v053_aged_running_upgrade"; + store.enqueue_batch(&pool, queue, 4, 1).await.unwrap(); + sqlx::query( + "UPDATE awa_queue_storage.ready_entries \ + SET run_at = now() - interval '5 minutes' WHERE queue = $1", + ) + .bind(queue) + .execute(&pool) + .await + .unwrap(); + sqlx::query( + "UPDATE awa_queue_storage.ready_segments \ + SET first_run_at = now() - interval '5 minutes' WHERE queue = $1", + ) + .bind(queue) + .execute(&pool) + .await + .unwrap(); + let claimed = store + .claim_runtime_batch_with_aging_for_instance( + &pool, + queue, + 1, + std::time::Duration::ZERO, + std::time::Duration::from_secs(60), + Uuid::new_v4(), + 4, + std::time::Duration::from_secs(30), + std::time::Duration::from_secs(1), + ) + .await + .expect("claim with the v052 storage protocol"); + assert_eq!(claimed.len(), 1); + assert!(claimed[0].claim.lease_claim_receipt); + assert_eq!(claimed[0].job.priority, 1); + let before: i64 = sqlx::query_scalar("SELECT count(*) FROM awa.jobs WHERE queue = $1") + .bind(queue) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(before, 0, "v052 loses the aged running claim"); + let ledger_before: serde_json::Value = sqlx::query_scalar( + "SELECT to_jsonb(batch) FROM awa_queue_storage.lease_claim_batches AS batch \ + WHERE queue = $1", + ) + .bind(queue) + .fetch_one(&pool) + .await + .unwrap(); + + migrations::run(&pool).await.expect("upgrade v052 to v053"); + assert_eq!( + migrations::current_version(&pool).await.unwrap(), + migrations::CURRENT_VERSION + ); + for _ in 0..2 { + let rows: Vec<(i64, i64, i16, serde_json::Value)> = sqlx::query_as( + "SELECT id, run_lease, attempt, args FROM awa.jobs \ + WHERE queue = $1 AND state = 'running'", + ) + .bind(queue) + .fetch_all(&pool) + .await + .unwrap(); + assert_eq!( + rows, + vec![( + claimed[0].job.id, + claimed[0].job.run_lease, + claimed[0].job.attempt, + claimed[0].job.args.clone(), + )], + "upgrading must expose exactly the existing running attempt" + ); + let ledger_after: serde_json::Value = sqlx::query_scalar( + "SELECT to_jsonb(batch) FROM awa_queue_storage.lease_claim_batches AS batch \ + WHERE queue = $1", + ) + .bind(queue) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!( + ledger_after, ledger_before, + "the live claim must be unchanged" + ); + + // Exercise the exported bundle too, including a repeated application. + let mut tx = pool.begin().await.unwrap(); + for (_, _, sql) in migrations::migration_sql_range(52, 53) { + sqlx::raw_sql(audited_sql(sql)) + .execute(tx.as_mut()) + .await + .unwrap(); + } + tx.commit().await.unwrap(); + } + assert_eq!( + store + .complete_runtime_batch(&pool, &claimed) + .await + .expect("the pre-upgrade claim must still complete"), + vec![(claimed[0].job.id, claimed[0].job.run_lease)] + ); +} + // ── Idempotency ────────────────────────────────────────────────── #[tokio::test] diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index 69453d02f..009d78f35 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -17690,6 +17690,211 @@ async fn test_jobs_compat_view_reports_receipt_running_claims() { } } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn test_jobs_compat_view_reports_aged_receipt_claims() { + jobs_compat_view_reports_aged_claims(CompatClaimMode::Batch).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn test_jobs_compat_view_reports_aged_materialized_leases() { + jobs_compat_view_reports_aged_claims(CompatClaimMode::Materialized).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn test_jobs_compat_view_reports_aged_row_local_claims() { + jobs_compat_view_reports_aged_claims(CompatClaimMode::RowLocal).await; +} + +enum CompatClaimMode { + Batch, + RowLocal, + Materialized, +} + +async fn jobs_compat_view_reports_aged_claims(mode: CompatClaimMode) { + let (_db_guard, pool) = setup_pool(6).await; + let queue = "qs_compat_aged_running"; + let schema = "awa_qs_compat_aged_running"; + let store = create_store_with_config( + &pool, + QueueStorageConfig { + schema: schema.to_string(), + queue_slot_count: 4, + lease_slot_count: 2, + claim_slot_count: 2, + lease_claim_receipts: !matches!(mode, CompatClaimMode::Materialized), + ..Default::default() + }, + ) + .await; + let args = serde_json::json!({ "id": 901 }); + let job_id = enqueue_job( + &pool, + &store, + &CompleteJob { id: 901 }, + InsertOpts { + queue: queue.to_string(), + priority: 4, + ..Default::default() + }, + ) + .await; + backdate_one_lane_ready_job_run_at( + &pool, + schema, + job_id, + Utc::now() - chrono::Duration::minutes(5), + ) + .await; + // Reprioritization retains the old ready row, tombstoned, with the same + // job and attempt identity. It must not become another running result. + assert!(store.set_priority(&pool, job_id, 3).await.unwrap()); + let claimed = store + .claim_runtime_batch_with_aging_for_instance( + &pool, + queue, + 1, + Duration::ZERO, + Duration::from_secs(60), + Uuid::new_v4(), + 4, + Duration::from_secs(30), + Duration::from_secs(1), + ) + .await + .expect("claim reprioritized job with aging"); + assert_eq!(claimed.len(), 1); + assert_eq!(claimed[0].job.id, job_id); + assert_eq!(claimed[0].job.priority, 1); + assert_eq!(claimed[0].claim.priority, 3); + + if matches!(mode, CompatClaimMode::RowLocal) { + // Represent the real claim in the legacy row-local receipt format. + // Keep its identity, attempt, claim time and deadline unchanged. + let converted = sqlx::query(audited_sql(format!( + r#" + WITH batches AS ( + DELETE FROM {schema}.lease_claim_batches + WHERE queue = $1 + RETURNING * + ) + INSERT INTO {schema}.lease_claims ( + claim_slot, receipt_id, job_id, run_lease, ready_slot, + ready_generation, queue, priority, attempt, max_attempts, + lane_seq, enqueue_shard, claimed_at, deadline_at + ) + SELECT batches.claim_slot, items.receipt_id, items.job_id, + items.run_lease, batches.ready_slot, batches.ready_generation, + batches.queue, batches.priority, items.attempt, + items.max_attempts, items.lane_seq, batches.enqueue_shard, + batches.claimed_at, batches.deadline_at + FROM batches + CROSS JOIN LATERAL unnest( + batches.job_ids, batches.run_leases, batches.receipt_ids, + batches.lane_seqs, batches.attempts, batches.max_attempts + ) AS items(job_id, run_lease, receipt_id, lane_seq, attempt, max_attempts) + "# + ))) + .bind(queue) + .execute(&pool) + .await + .expect("convert the claim to the row-local receipt format"); + assert_eq!(converted.rows_affected(), 1); + } + + type Projection = (i64, String, i64, i16, serde_json::Value, i16); + async fn read_projection(pool: &sqlx::PgPool, job_id: i64) -> Vec { + sqlx::query_as( + "SELECT id, state::text, run_lease, attempt, args, priority \ + FROM awa.jobs WHERE id = $1 ORDER BY run_lease", + ) + .bind(job_id) + .fetch_all(pool) + .await + .expect("read aged job from SQL compatibility view") + } + assert_eq!( + read_projection(&pool, job_id).await, + vec![( + job_id, + "running".to_string(), + claimed[0].job.run_lease, + claimed[0].job.attempt, + args.clone(), + claimed[0].job.priority, + )], + "aging must not hide a live claim or duplicate a tombstoned ready row" + ); + + let retried = store + .retry_after( + &pool, + job_id, + claimed[0].job.run_lease, + Duration::ZERO, + None, + ) + .await + .expect("retry aged claim") + .expect("current attempt must be retryable"); + assert_eq!( + read_projection(&pool, job_id).await, + vec![( + job_id, + "retryable".to_string(), + claimed[0].job.run_lease, + claimed[0].job.attempt, + args.clone(), + retried.priority, + )], + "closed receipts must not remain visible as running" + ); + assert_eq!( + store + .promote_due(&pool, JobState::Retryable, 1) + .await + .expect("promote retried job"), + 1 + ); + let reclaimed = store + .claim_runtime_batch(&pool, queue, 1, Duration::from_secs(30)) + .await + .expect("claim the next attempt"); + assert_eq!(reclaimed.len(), 1); + assert!(reclaimed[0].job.run_lease > claimed[0].job.run_lease); + assert_eq!( + read_projection(&pool, job_id).await, + vec![( + job_id, + "running".to_string(), + reclaimed[0].job.run_lease, + reclaimed[0].job.attempt, + args.clone(), + reclaimed[0].job.priority, + )], + "retained ready rows from older attempts must not duplicate the current one" + ); + assert_eq!( + store + .complete_runtime_batch(&pool, &reclaimed) + .await + .expect("complete current attempt"), + vec![(job_id, reclaimed[0].job.run_lease)] + ); + assert_eq!( + read_projection(&pool, job_id).await, + vec![( + job_id, + "completed".to_string(), + reclaimed[0].job.run_lease, + reclaimed[0].job.attempt, + args, + reclaimed[0].job.priority, + )], + "completed attempts must not also remain visible as running" + ); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn test_queue_storage_cursor_advance_requires_claim_evidence() { cursor_advance_requires_claim_evidence(true).await;