diff --git a/CHANGELOG.md b/CHANGELOG.md index 6c4e7013..ac25f35b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,8 @@ Notable changes between releases. Detailed migration notes for storage transitio ### Fixed +- **`awa.jobs` is receipt-aware and no longer pathologically slow under backlog ([#422](https://github.com/hardbyte/awa/issues/422), migration v044).** The SQL-compat view had two defects found during the 0.6.1→0.7 upgrade rehearsal. *Performance:* the available-branch filter spelled `lane_seq >= sequence_next_value(claims.seq_name)` inline against the partitioned `ready_entries` scan; `sequence_next_value` is a VOLATILE PL/pgSQL function, so the planner could neither use the predicate as an index bound nor cache it across rows — every ready child was scanned in full and each surviving row paid a dynamic-SQL catalog round-trip (measured ~90ms → ~6ms for the branch on a 20k-row backlog; the gap grows with un-pruned sealed generations). v044 materializes the per-lane cursors once per statement — the same shape the claim path has used since v027/v039 — and reading each cursor once is also a more consistent snapshot than per-row evaluation. *Correctness:* receipt claims that never materialized into `leases` (row-local `lease_claims`, and by default compact `lease_claim_batches`) were invisible, so `SELECT state, count(*) FROM awa.jobs` reported `running=0` while workers held live work. The view now carries the open-receipt legs the admin surface shipped in #410 (`admin::state_counts` / `queue_storage::open_receipt_running_claims_sql`), projecting the claim ledger's own attempt / claim-time / deadline values and anti-joined against closures, closure batches, materialised leases, and terminal/deferred/DLQ supersession — a job reports at most one state outside the brief post-commit cursor-lag window the claim protocol itself has (identical to the admin surfaces; it self-heals on the lane's next claim). Rows previously returned are unchanged; the running rows are additive. **No 0.6.x backport is required:** released 0.6.x workers have no schema-version gate at startup, so both upgrade orderings keep working — only a ≤0.6.x *migrator* refuses via the #392 fail-safe, which is correct (past-v043 migrations are applied by the 0.7 binary). Verified by the compat matrix against pinned release artifacts (`scripts/compat-matrix.sh` now runs a `forward-0.6.6` leg alongside 0.6.2/0.6.0/0.5.7) plus a released-binary enqueue→claim→complete rehearsal on a v044 database. + - **`awa migrate --extract-to` wrote only the first 15 migrations.** The target path was built by substituting the migration description into the filename, so v017 — whose description contains `/` — resolved to a nested directory that does not exist. The command aborted there, leaving a partial extraction that looked plausible but silently omitted two thirds of the schema, breaking the documented external-runner workflow for every range reaching v017. Only path separators are now substituted, so every filename this tool has already published stays byte-identical — re-extracting into an existing directory cannot leave two files for the same version. `V17` and `V21` are the only names that change, and neither could previously be written at all. All paths are computed and checked for collisions before anything is written, and a write failure names the file. A test asserts every migration is extracted with byte-identical SQL. - **`python -m awa migrate --sql` matches the Rust CLI's transaction wrapper.** The two CLIs are documented as interchangeable, so the Python one emitted unwrapped SQL for the same `| psql` workflow. It now renders the identical wrapper and accepts the same `--no-transaction` escape hatch, taking the lock key from the new `awa.migration_lock_key()` binding rather than a duplicated literal. diff --git a/awa-model/migrations/v044_jobs_compat_receipt_running.sql b/awa-model/migrations/v044_jobs_compat_receipt_running.sql new file mode 100644 index 00000000..f393fb9f --- /dev/null +++ b/awa-model/migrations/v044_jobs_compat_receipt_running.sql @@ -0,0 +1,577 @@ +-- v044: refresh awa.jobs_compat() (#422). +-- +-- Two changes to the SQL-compat projection: +-- +-- 1. Materialize the per-lane claim cursors once per statement. The previous +-- body filtered ready rows with +-- `lane_seq >= sequence_next_value(claims.seq_name)` inline against the +-- partitioned ready_entries scan. `sequence_next_value` is a VOLATILE +-- PL/pgSQL function (a dynamic-SQL sequence read), so the planner could +-- neither use the predicate as an index bound nor cache it across rows: +-- every ready_entries child was scanned in full and each surviving row +-- paid a catalog round-trip (~15x slower on a 20k-row backlog; the gap +-- grows with un-pruned sealed generations). The claim path has computed +-- cursor values once per call since v027/v039; this brings the read +-- surface in line. Reading each cursor once per statement is also a more +-- consistent snapshot than per-row reads, which could observe a cursor +-- advanced by another transaction mid-scan. +-- +-- 2. Surface receipt-plane running claims (#422 correctness half). Claims +-- that have not materialised into `leases` -- legacy row-local +-- `lease_claims` and default compact `lease_claim_batches` members -- +-- were invisible: `SELECT state, count(*) FROM awa.jobs` reported +-- running=0 while workers held live work. This ports the open-receipt +-- shape the admin surface shipped in #410 (admin::state_counts / +-- queue_counts_exact / queue_storage::open_receipt_running_claims_sql) +-- into the full-row view, projecting the claim ledger's own attempt / +-- claimed-at / deadline values so running rows read the same as +-- materialised leases do. The anti-join set against durable closures, +-- closure batches, materialised leases, and terminal/deferred/DLQ +-- supersession means a job reports at most one state outside the brief +-- post-commit cursor-lag window the claim protocol itself has (receipt +-- evidence commits before the lane's claim cursor advances — identical +-- to the admin surfaces; it self-heals on the lane's next claim). Like +-- the admin surface, the running legs join ready_entries for the job +-- body, so an open claim whose ready row were missing would not be +-- listed; prune guards (`queue_prune_has_unclosed_claim_refs_tx`) make +-- that unreachable. +-- +-- N-1 compatibility: this is a server-side refresh of an awa-schema +-- function and view. No runtime parses the body; binaries of any version +-- that query `awa.jobs` observe the new rows through the refreshed +-- definition immediately. Rows previously returned are unchanged (same +-- columns, same available/deferred/terminal/DLQ membership); the change is +-- additive running rows plus faster planning. +-- +-- No 0.6.x backport is required. Released 0.6.x workers carry no +-- schema-version gate at startup (queue_storage_schema_ready checks objects, +-- not versions), so they keep running against a v044 database in both +-- upgrade orderings; only a <=0.6.x *migrator* refuses via the #392 +-- fail-safe ("binary too old for the database schema"), which is the +-- designed behavior -- migrations past v043 are applied by the 0.7 binary. +-- Verified against the released artifacts by the compat matrix +-- (scripts/compat-matrix.sh: forward-0.6.6 / forward-0.6.2 / forward-0.6.0 / +-- forward-0.5.7 lifecycle legs on a v044 schema, plus the finalized-upgrade +-- backward leg) and a released-binary rehearsal. +-- +-- Re-runnable: CREATE OR REPLACE only; no data changes. Transaction-safe. + +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.priority = claims.priority + AND ready.enqueue_shard = claims.enqueue_shard + AND ready.lane_seq = claims.lane_seq + AND ready.job_id = claims.job_id + 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.priority = batches.priority + AND ready.enqueue_shard = batches.enqueue_shard + AND ready.lane_seq = items.lane_seq + AND ready.job_id = items.job_id + 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.priority = leases.priority + AND ready.enqueue_shard = leases.enqueue_shard + AND ready.lane_seq = leases.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 (44, 'Refresh jobs_compat(): materialized claim cursors and receipt-plane running rows (#422)') +ON CONFLICT (version) DO NOTHING; diff --git a/awa-model/src/migrations.rs b/awa-model/src/migrations.rs index 5dcab06e..017aa744 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 = 43; +pub const CURRENT_VERSION: i32 = 44; /// Migrations that require an exclusive (no-live-runtime) upgrade window. /// @@ -256,6 +256,11 @@ const MIGRATIONS: &[(i32, &str, &[&str])] = &[ "Append-only ring-rotation ledgers and terminal-rollup deltas (#371)", &[V18_UP, V23_UP, V43_UP], ), + ( + 44, + "Refresh jobs_compat(): materialized claim cursors and receipt-plane running rows (#422)", + &[V44_UP], + ), ]; const V1_UP: &str = include_str!("../migrations/v001_canonical_schema.sql"); @@ -300,6 +305,7 @@ const V40_UP: &str = include_str!("../migrations/v040_finalize_with_drain_runtim const V41_UP: &str = include_str!("../migrations/v041_queue_runtime_overrides.sql"); const V42_UP: &str = include_str!("../migrations/v042_compact_deadline_claims.sql"); const V43_UP: &str = include_str!("../migrations/v043_ring_rotation_ledger.sql"); +const V44_UP: &str = include_str!("../migrations/v044_jobs_compat_receipt_running.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/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index 558acd2c..83309dcb 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -14908,3 +14908,103 @@ async fn test_queue_storage_lowering_enqueue_shards_drains_existing_rows() { client.shutdown(Duration::from_secs(5)).await; } + +/// The migration-owned `awa.jobs` compat view must surface receipt-plane +/// running claims (`#422`): jobs held by open `lease_claim_batches` (and +/// legacy row-local `lease_claims`) previously vanished from the view +/// entirely — running=0 while workers held live work. The view also computes +/// per-lane claim cursors once per statement (v044) instead of evaluating the +/// volatile `sequence_next_value` per ready row, which is what made the view +/// pathologically slow under an un-pruned backlog. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn test_jobs_compat_view_reports_receipt_running_claims() { + let (_db_guard, pool) = setup_pool(6).await; + let queue = "qs_compat_receipt_running"; + let schema = "awa_qs_compat_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: true, + ..Default::default() + }, + ) + .await; + + for id in 0..3 { + enqueue_job( + &pool, + &store, + &CompleteJob { id }, + InsertOpts { + queue: queue.to_string(), + ..Default::default() + }, + ) + .await; + } + + async fn read_state_counts(pool: &sqlx::PgPool, queue: &str) -> Vec<(String, i64)> { + sqlx::query_as::<_, (String, i64)>( + "SELECT state::text, count(*)::bigint FROM awa.jobs WHERE queue = $1 GROUP BY state", + ) + .bind(queue) + .fetch_all(pool) + .await + .expect("read compat-view states") + } + + let baseline = read_state_counts(&pool, queue).await; + assert_eq!( + baseline, + vec![("available".to_string(), 3)], + "before claiming, all enqueued jobs must be available and nothing else may appear" + ); + + // Claim through the runtime batch API so the claim cursor advances + // post-commit exactly as a real worker would; receipts mode records the + // claims as open compact batches rather than materialised leases. + let claimed = store + .claim_runtime_batch(&pool, queue, 2, Duration::from_secs(30)) + .await + .expect("claim two jobs into the receipt plane"); + assert_eq!(claimed.len(), 2, "two jobs must be claimed"); + + let after = read_state_counts(&pool, queue).await; + let mut states: Vec<(String, i64)> = after; + states.sort(); + assert_eq!( + states, + vec![ + ("available".to_string(), 1), + ("running".to_string(), 2), + ], + "open receipt claims must be visible as running (#422), with the unclaimed remainder available" + ); + + // Running rows must read from the claim ledger, not the pre-claim ready + // row: attempt is the incremented claim attempt, attempted_at is the + // claim time, and the 30s claim deadline surfaces as deadline_at. + type RunningProjection = (i16, Option>, Option>); + let running_rows: Vec = sqlx::query_as( + "SELECT attempt, attempted_at, deadline_at FROM awa.jobs \ + WHERE queue = $1 AND state = 'running' ORDER BY id", + ) + .bind(queue) + .fetch_all(&pool) + .await + .expect("read running-row projections"); + assert_eq!(running_rows.len(), 2); + for (attempt, attempted_at, deadline_at) in &running_rows { + assert_eq!(*attempt, 1, "running rows must project the claimed attempt"); + let claimed_at = attempted_at.expect("running rows must expose the claim time"); + let deadline = deadline_at.expect("the 30s claim deadline must surface on running rows"); + assert!( + deadline > claimed_at, + "deadline_at must be the claim time plus the claim deadline" + ); + } +} diff --git a/correctness/storage/MAPPING.md b/correctness/storage/MAPPING.md index 86f0fd01..c8d6ff62 100644 --- a/correctness/storage/MAPPING.md +++ b/correctness/storage/MAPPING.md @@ -99,7 +99,7 @@ The TLA+ lifecycle model does not represent the completed-history rollup cache, `AwaSegmentedStorage.tla` models storage state, not every SQL projection over that state. Public and admin reads are therefore refinement obligations on the SQL implementation: -- `awa.jobs` / `awa.jobs_compat()` is the compatibility view used by SQL adapters and operational queries. Migration `awa-model/migrations/v028_ready_tombstones.sql` keeps queue-storage available rows shard-aware, uses the sequence-backed claim cursor, and skips `ready_tombstones`; lease-backed rows join ready bodies by `(queue, priority, enqueue_shard, lane_seq)`. +- `awa.jobs` / `awa.jobs_compat()` is the compatibility view used by SQL adapters and operational queries. Migration `awa-model/migrations/v028_ready_tombstones.sql` keeps queue-storage available rows shard-aware, uses the sequence-backed claim cursor, and skips `ready_tombstones`; lease-backed rows join ready bodies by `(queue, priority, enqueue_shard, lane_seq)`. Migration `v044_jobs_compat_receipt_running.sql` materializes the per-lane claim cursors once per statement (the inline VOLATILE `sequence_next_value` filter defeated index bounds and re-evaluated per row) and adds the open-receipt running legs (row-local `lease_claims` + compact `lease_claim_batches`, anti-joined against closures/closure-batches/leases/terminal/deferred/DLQ), matching the admin read side (#410, #422). - `awa-model/src/admin.rs::queue_storage_current_jobs_cte`, `awa-model/src/admin.rs::state_counts`, and `awa-worker/src/client.rs::health_check` are the Rust read-side equivalents. They must preserve the same enqueue-shard predicates or multi-shard queues can overcount available rows or hydrate a row from the wrong shard. - `test_queue_storage_multi_shard_public_available_counts_are_exact` is the code-level regression test for this projection boundary. A future TLA+ projection model could make this formal by deriving `JobsCompatAvailable` from the storage variables and asserting it equals `CurrentReady`; the current storage model stops at the underlying lifecycle and prune state. diff --git a/docs/0.7-roadmap.md b/docs/0.7-roadmap.md index 09f118c3..f6b540db 100644 --- a/docs/0.7-roadmap.md +++ b/docs/0.7-roadmap.md @@ -241,7 +241,7 @@ this roadmap and not yet filed. | 036 | Public surface stability policy | Accepted | #369 / D6 | | 037 | Canonical engine deprecation & removal | Accepted | #370 / D2 | | 042 | Caller-owned finalization transactions | Accepted | #401 / #342 | -| 03x | Segment/cursor storage evolution | Draft from E1/E3; status decided at Gate A | #295 | +| 044 | Segment/cursor storage evolution | Gate A decided 2026-08-22: RFC graduates to 0.8 with evidence; allocator ideas shipped in place via v027–v043 | #295 | --- diff --git a/docs/adr/044-storage-evolution-gate-a.md b/docs/adr/044-storage-evolution-gate-a.md new file mode 100644 index 00000000..82920ead --- /dev/null +++ b/docs/adr/044-storage-evolution-gate-a.md @@ -0,0 +1,147 @@ +# ADR-044: Gate A — storage evolution for 0.7; the segment-engine RFC graduates to 0.8 + +## Status + +Accepted — decided 2026-08-22 against the evidence listed below. This is the +**Gate A** scope decision defined in the [0.7 roadmap §5](../0.7-roadmap.md) +and required by the [#383](https://github.com/hardbyte/awa/issues/383) +performance contract ("Record the #295 Gate A decision"). It resolves +[#295](https://github.com/hardbyte/awa/issues/295) for the 0.7 scope: no +segment-storage restructuring migrations enter 0.7; the RFC graduates to 0.8 +with its evidence attached. + +## Context + +[#295](https://github.com/hardbyte/awa/issues/295) proposes replacing +queue_storage's mutable lifecycle shape with append-only rotation segments plus +a cursor allocator, motivated by the #169 finding that a pinned MVCC horizon +degraded a 0.6-era engine from 799 → 387 jobs/s at 800 jobs/s offered over a +2-hour pin. The roadmap's decision D1 defaulted to *evolution of the existing +engine via staged migrations* and made that default reversible only by Gate A: +restructuring migrations enter 0.7 **only if all three** hold: + +1. a prototype is ≥ parity clean-phase and strictly better at the 1,600/s + pinned-horizon shape; +2. ≥40% of WAL/job is attributable to removable architecture (E3), i.e. the + redesign has proven headroom; +3. the change is deliverable as staged in-place migrations (per D1) with TLA+ + deltas identified. + +Since that rule was written, the 0.7 performance campaign landed the allocator +ideas *inside* the current engine as staged migrations, and measured them: + +- **v027/v039** — sequence-backed lane cursors and cache-free ready-segment + routing: the claim path walks the `ready_segments` control plane ordered by + `next_lane_seq` with an index short-circuit at `LIMIT 1` (the "cursor + allocator" of the RFC, in situ). +- **#409 / v043 (#371)** — idle rotation skip, then append-only ring-rotation + ledgers: the ring singletons stopped being hot mutable rows entirely. + Pinned-horizon soak: dead tuples flat at ~6 versus 145–298 accrual on the + pre-ledger intermediate ([2026-07-11 gate](https://github.com/hardbyte/postgresql-job-queue-benchmarking/tree/main/results/2026-07-11-awa-07-alpha-gate)). +- **#410 / v042** — compact deadline receipt claims removed ~83k steady-state + live `lease_claims` rows at 256-worker saturation (the E2/#246 fix). +- **Release-candidate cells (2026-08-22)** — main @ `8c27951` vs v0.6.0 at + W=256 depth-target saturation: **11,945/s @ p99 317 ms vs 10,568/s @ p99 + 532 ms** (+13% throughput, −40% tail), ref800 parity at p99 21 ms. +- **E3 attribution** (recorded on #415): ~52% of WAL bytes at saturation are + B-tree/index maintenance — above the 40% headroom threshold. The landed work + did **not** reduce it (WAL byte-parity with v0.6.0 per 5k cell: 1,142 vs + 1,143 MB); E9.4b confirmed BRIN cannot serve the ordered-LIMIT claim + contract, so no in-place migration captures this share. + +## Decision + +**Gate A resolves to evolution: no segment-storage restructuring migrations +enter 0.7.** The RFC's remaining scope graduates to 0.8 with this evidence +attached. Concretely: + +1. **Criteria (i) fails as specified, and its question is answered anyway.** + No side-by-side prototype (P-b/P-c) was built. Instead the strongest + candidate allocator shape — segments + cursors + append-only ledgers — was + implemented inside the engine behind six individually-tested migrations and + measured better at every recorded shape. The motivating degradation + mechanism (dead-tuple accumulation on hot control rows under a pinned MVCC + horizon) is eliminated at the representation level, not mitigated. +2. **Criterion (ii) passes the threshold but not the delivery test in (iii).** + The ~52% index-maintenance WAL share is real headroom, but capturing it + requires changing what the lifecycle rows *are* (index-avoiding segment + storage), which D1 correctly prices as a third engine identity or an + expand→flip→contract lifecycle swap of the core tables — exactly the cost + 0.5→0.6 paid, with restore-only rollback. No staged-migration path to that + reduction has been identified (BRIN rejected; the composite lane indexes + are load-bearing for the ordered-LIMIT claim contract). +3. **Criterion (iii) held for everything actually shipped** — v016 through + v043 restructured cursors, routing, receipts, terminal history, and ring + bookkeeping in place, each individually benchmarked, without a new engine + identity. That delivery record is itself evidence the engine absorbs + structural change; it does not evidence a ceiling. + +What 0.7 ships instead of restructuring: the landed stack above, plus the E9 +deployment guidance (`wal_compression=lz4`, leave `commit_delay` at 0, do not +pin `plan_cache_mode`) routed to the operations handbook (#379) and +`awa doctor` advisories (#373). + +## Answers to the RFC's five questions + +1. **Claim allocator** — answered by construction: per-lane sequence cursors + (`queue_claim_heads.seq_name`), a non-overlapping ready-segment control + plane for O(1) routing, `FOR UPDATE SKIP LOCKED` on the head row for + fairness, and append-only rotation ledgers so the allocator's bookkeeping + is vacuum-cold under any MVCC horizon. Per-row retries/heartbeats/ + cancellation compose unchanged (ADR-023 receipt plane, ADR-003 rescue). +2. **Receipt plane integration** — survived and strengthened: ring-partitioned + receipts (ADR-023) now write compact batch claims (v038) and compact + deadline claims (v042); terminal history folds into narrow rollups + (ADR-026 + v043 deltas). No redesign needed to keep them. +3. **Migration story** — the staged expand→flip→contract pattern with a + released stepping-stone (ADR-037, ADR-040, ADR-041) is now proven twice. + A third engine identity would repeat the 0.5→0.6 operator cost without a + measured win; if 0.8 takes up the segment design, it should reuse this + machinery rather than a hard cutoff. +4. **Comparable designs** — pgque holds flat under pinned horizons by trading + away per-job state. Awa kept the full job-queue contract and removed the + degradation mechanism (hot-row churn) surgically; the residual WAL gap vs + pgque (~2 KiB/job) is dominated by contract evidence — per-attempt identity, + receipts, terminal batches — plus index maintenance, not by avoidable + bookkeeping. +5. **TLA+ coverage** — the ledger migration updated `AwaStorageLockOrder`, + `AwaDeadTupleContract` (ledgers modeled as cold `RowVacuum` with + horizon-gated folds), added the `MixedFleet` staged-upgrade invariant, and + TLC caught a real authority-read TOCTOU during #371. A segment redesign + re-models all of this; that cost belongs to whatever release takes it. + +## Consequences + +- 0.7's performance story rests on measured parity-or-better cells against + v0.6.0 and the elimination of the pinned-MVCC degradation mechanism, not on + a storage replacement. +- The known residual ceilings are explicit and tracked: **#418** (claim yield + capped at one generation per call under extreme fragmentation — latent, + requires deeper fragmentation than bench shapes produce; the candidate fix + is a claim-CTE change, deliverable in place) and the **~52% WAL share** of + index maintenance (documented as the price of the claim contract until an + index-avoiding shape exists). +- The 60-minute pinned-MVCC reference soak in **ledger authority** remains an + open #383 performance-contract item for the release candidate; it validates + the shipped design and does not gate this decision (its mechanism-level + evidence is already recorded). + +### Reversal conditions (what reopens the redesign for 0.8) + +Any of these, evidenced on released artifacts, reopens #295 with priority: + +- sustained WAL-flush-bound ceilings at documented operator shapes after the + tuning presets ship; +- #418-class fragmentation wedges observed in a real fleet (not only bench + shapes); +- long-horizon latency drift or bloat reappearing in ledger authority during + the pinned-MVCC soak or nightly chaos runs. + +## References + +- [#295](https://github.com/hardbyte/awa/issues/295) — the RFC this gate closes for 0.7 +- [#383](https://github.com/hardbyte/awa/issues/383) — performance contract naming Gate A +- [0.7 roadmap](../0.7-roadmap.md) — D1, §5 experiments, Gate A rule +- [#169 spike](../archive/0.6-storage-design/issue-169-storage-spike.md) — original degradation evidence +- [bench repo 2026-07-11 gate](https://github.com/hardbyte/postgresql-job-queue-benchmarking/tree/main/results/2026-07-11-awa-07-alpha-gate) + and 2026-08-22 RC cells — measured comparisons cited above diff --git a/docs/adr/README.md b/docs/adr/README.md index 9bbae615..e0f7b7a9 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -51,6 +51,7 @@ Each record preserves its status, context, decision, consequences, and alternati | 041 | [Rolling upgrade policy](041-rolling-upgrade-policy.md) | Rolling upgrades use expand → capability-gated flip → later contract; version floors guard expand migrations, while database fences and real N-1 rehearsals guard irreversible flips. | | | 042 | [Caller-owned finalization transactions](042-caller-owned-finalization-transactions.md) | A distinct handler type commits application rows and exact-lease completion in one transaction through a least-privilege finalization function (#401). | | | 043 | [PostgreSQL capability functions and least-privilege runtime roles](043-postgresql-capability-functions.md) | Replace blanket runtime table/function grants with allowlisted, role-specific capability entry points owned by a bounded execution role (#452); blanket definer conversion is rejected. | Proposed | +| 044 | [Gate A — storage evolution for 0.7](044-storage-evolution-gate-a.md) | The #295 segment-engine RFC graduates to 0.8: the allocator ideas landed inside the engine as staged migrations and measured better; the remaining WAL headroom has no in-place delivery path (#295, #383). | | ## Correctness evidence diff --git a/scripts/compat-matrix.sh b/scripts/compat-matrix.sh index 89d54038..d94109d0 100755 --- a/scripts/compat-matrix.sh +++ b/scripts/compat-matrix.sh @@ -6,6 +6,12 @@ # those skews with PINNED RELEASE ARTIFACTS (awa-pg wheels from PyPI — the # compiled runtime, not a source build of an old tag): # +# forward-0.6.6 latest released 0.6.x lifecycle (enqueue/claim/complete/ +# cancel) against the newest schema. Proves additive refresh +# migrations past the 0.6 migrator's recognized ceiling +# (v043, e.g. the #422 jobs_compat refresh) do not need a +# 0.6.x patch: workers carry no version gate, only the +# migrator does. # forward-0.6.2 supported N-1 lifecycle (enqueue/claim/complete/cancel) # against the newest schema in columns authority, followed # by a post-flip fence check. @@ -88,11 +94,17 @@ SQL echo "── setup: pinned release artifacts (PyPI wheels)" uv venv --quiet --clear .compat-venv-060 uv pip install --quiet --python .compat-venv-060 "awa-pg==0.6.0" +uv venv --quiet --clear .compat-venv-066 +uv pip install --quiet --python .compat-venv-066 "awa-pg==0.6.6" uv venv --quiet --clear .compat-venv-062 uv pip install --quiet --python .compat-venv-062 "awa-pg==0.6.2" uv venv --quiet --clear .compat-venv-057 uv pip install --quiet --python .compat-venv-057 "awa-pg==0.5.7" +echo "── leg: forward-0.6.6 (latest released 0.6.x lifecycle on newest schema)" +DATABASE_URL="${BASE_URL}/${FWD_DB}" COMPAT_VERSION=0.6.6 COMPAT_QUEUE=compat_forward_066 \ + .compat-venv-066/bin/python "${SCRIPT_DIR}/compat/forward_060.py" + echo "── leg: forward-0.6.2 (supported N-1 lifecycle on newest schema)" DATABASE_URL="${BASE_URL}/${FWD_DB}" COMPAT_VERSION=0.6.2 COMPAT_QUEUE=compat_forward_062 \ .compat-venv-062/bin/python "${SCRIPT_DIR}/compat/forward_060.py"