From 515b78002ed5b860a9c0d9c5d027dc63591b7a34 Mon Sep 17 00:00:00 2001 From: chencaizhi Date: Thu, 29 Oct 2026 21:07:46 +0800 Subject: [PATCH 1/2] fix: keep aged running jobs visible in SQL compat view Closes hardbyte/awa#569 --- CHANGELOG.md | 9 + .../v053_jobs_compat_priority_aging.sql | 564 ++++++++++++++++++ awa-model/src/migrations.rs | 8 +- awa/tests/migration_test.rs | 125 ++++ awa/tests/queue_storage_runtime_test.rs | 201 +++++++ 5 files changed, 906 insertions(+), 1 deletion(-) create mode 100644 awa-model/migrations/v053_jobs_compat_priority_aging.sql diff --git a/CHANGELOG.md b/CHANGELOG.md index 54759f505..78df41409 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,15 @@ 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. + 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..478ae5e6d --- /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, + ready.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, + ready.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..58356fa29 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -17690,6 +17690,207 @@ 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); + async fn read_projection(pool: &sqlx::PgPool, job_id: i64) -> Vec { + sqlx::query_as( + "SELECT id, state::text, run_lease, attempt, args \ + 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(), + )], + "aging must not hide a live claim or duplicate a tombstoned ready row" + ); + + 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(), + )], + "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(), + )], + "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, + )], + "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; From 32aa6bc2c7a9b09a8ee8dacf2e8964b206429661 Mon Sep 17 00:00:00 2001 From: chencaizhi Date: Thu, 29 Oct 2026 21:53:11 +0800 Subject: [PATCH 2/2] fix: preserve effective priority in receipt projections --- CHANGELOG.md | 2 ++ .../migrations/v053_jobs_compat_priority_aging.sql | 4 ++-- awa/tests/queue_storage_runtime_test.rs | 10 +++++++--- 3 files changed, 11 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 78df41409..820d8d60e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,8 @@ Notable changes between releases. Detailed migration notes for storage transitio 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. diff --git a/awa-model/migrations/v053_jobs_compat_priority_aging.sql b/awa-model/migrations/v053_jobs_compat_priority_aging.sql index 478ae5e6d..3ecb405d7 100644 --- a/awa-model/migrations/v053_jobs_compat_priority_aging.sql +++ b/awa-model/migrations/v053_jobs_compat_priority_aging.sql @@ -226,7 +226,7 @@ BEGIN ready.queue, ready.args, 'running'::awa.job_state AS state, - ready.priority, + claims.priority, claims.attempt, claims.max_attempts, ready.run_at, @@ -310,7 +310,7 @@ BEGIN ready.queue, ready.args, 'running'::awa.job_state AS state, - ready.priority, + batches.priority, batches.attempts[items.ord], batches.max_attempts[items.ord], ready.run_at, diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index 58356fa29..009d78f35 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -17802,10 +17802,10 @@ async fn jobs_compat_view_reports_aged_claims(mode: CompatClaimMode) { assert_eq!(converted.rows_affected(), 1); } - type Projection = (i64, String, i64, i16, serde_json::Value); + 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 \ + "SELECT id, state::text, run_lease, attempt, args, priority \ FROM awa.jobs WHERE id = $1 ORDER BY run_lease", ) .bind(job_id) @@ -17821,11 +17821,12 @@ async fn jobs_compat_view_reports_aged_claims(mode: CompatClaimMode) { 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" ); - store + let retried = store .retry_after( &pool, job_id, @@ -17844,6 +17845,7 @@ async fn jobs_compat_view_reports_aged_claims(mode: CompatClaimMode) { claimed[0].job.run_lease, claimed[0].job.attempt, args.clone(), + retried.priority, )], "closed receipts must not remain visible as running" ); @@ -17868,6 +17870,7 @@ async fn jobs_compat_view_reports_aged_claims(mode: CompatClaimMode) { 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" ); @@ -17886,6 +17889,7 @@ async fn jobs_compat_view_reports_aged_claims(mode: CompatClaimMode) { reclaimed[0].job.run_lease, reclaimed[0].job.attempt, args, + reclaimed[0].job.priority, )], "completed attempts must not also remain visible as running" );