diff --git a/CHANGELOG.md b/CHANGELOG.md index c6fbeb9c0..dec026559 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ Notable changes between releases. Detailed migration notes for storage transitio ### Upgrading - Run `awa migrate` once as the schema owner to apply `prune_controls` after the existing `claim_batch_job_index` and `insert_jobs_compat_batch` patches. It adds two tables and functions without altering existing tables; schema stays v40. Upgrade or fence every possible maintenance leader before relying on halt. See the [0.6.15 to 0.6.16 upgrade guide](docs/upgrade-0.6.15-to-0.6.16.md). +- The same run re-applies `claim_batch_job_index` on a 0.6.15 database, replacing each schema's `ensure_claim_batch_job_index` function. Like the first application it builds no index and takes no table locks. External migration runners: replace `R__claim_batch_job_index.sql` with the new version as well. ### Added @@ -14,11 +15,13 @@ Notable changes between releases. Detailed migration notes for storage transitio ### Fixed +- Claim prune no longer treats a partial GIN index on `lease_claim_batches.job_ids`, or a session with `quote_all_identifiers` on, as the claim batch index. It recognises only a valid, non-partial GIN index on `job_ids` alone. When another index or relation already has the index's name, `ensure_claim_batch_job_index` leaves it in place, returns false and raises a warning naming it, instead of reporting the slot indexed; the slot keeps scanning until an operator drops or renames it. +- Refreshing `ensure_claim_batch_job_index` as a migration role that owns it but may only `SET ROLE` to the table owner (PostgreSQL 16+ membership `WITH INHERIT FALSE`) no longer leaves `EXECUTE` with `PUBLIC`. The refresh used to hand the function to the table owner first, after which that role could no longer revoke or grant on it, and PostgreSQL only warned. It now resets `EXECUTE` after the transfer, acting as the new owner when it cannot as itself, so exactly the roles holding `TRUNCATE` on `lease_claim_batches` can execute it, the former owner included. Migrations run with the owner as the effective role, as documented, were not affected. - `admin::get_job` and the other queue-storage point reads of one job no longer return `JobNotFound` for a failed job while queue prune moves its row out of the slot being pruned. Since 0.6.14 the read used one snapshot, and a prune that committed after that snapshot was taken left the old slot empty and the moved row invisible to it. Prune moves each retained failed row once per ring cycle, and on a new schema prunes slot 0 right after the first rotation, so a read polling for `failed` could hit it within a second of a worker starting. A read that finds nothing now reads once more in a new snapshot. ## [0.6.15] — 2026-11-09 -Recommended for everyone on 0.6.14 or 0.6.13 using queue storage. A worker whose completions or lifecycle hooks fall behind stops claiming instead of holding every finished job in memory until it runs out, by-job lookups of compact claims no longer read every retained claim batch, and large COPY batches hold a queue lane's enqueue lock only while they write that lane. +Recommended for everyone on 0.6.14 or 0.6.13 using queue storage. A worker whose completions or lifecycle hooks fall behind stops claiming instead of holding every finished job in memory until it runs out, by-job lookups of compact claims no longer read every retained claim batch, and large COPY batches take a queue lane's enqueue lock when they write that lane rather than at their first ready row. ### Upgrading @@ -47,7 +50,7 @@ Recommended for everyone on 0.6.14 or 0.6.13 using queue storage. A worker whose ### Changed -- Compatibility COPY inserts (`insert_many_copy` in Rust and Python) hold a queue lane's enqueue lock only while they write that lane, instead of from the batch's first ready row until commit. Many producers enqueuing large batches into one queue no longer queue behind each other's whole batches, and promotion into a busy queue waits less. The batch goes through the new `awa.insert_jobs_compat_batch` function, which prepares job ids, unique claims and payloads first, then registers each new lane and reserves each lane once, in the sorted `(queue, priority, shard)` order every other producer uses. Each call checks its connection for the function and keeps the per-row path where it is missing or not executable. Results are unchanged: input order, with jobs whose unique key is already claimed skipped. +- Compatibility COPY inserts (`insert_many_copy` in Rust and Python) take a queue lane's enqueue lock when the batch writes that lane, instead of at the batch's first ready row. The lock is transaction-scoped, so it is still held until the transaction ends. Many producers enqueuing large batches into one queue no longer queue behind each other's whole batches, and promotion into a busy queue waits less. The batch goes through the new `awa.insert_jobs_compat_batch` function, which prepares job ids, unique claims and payloads first, then registers each new lane and reserves each lane once, in the sorted `(queue, priority, shard)` order every other producer uses. Each call checks its connection for the function and keeps the per-row path where it is missing or not executable. Results are unchanged: input order, with jobs whose unique key is already claimed skipped. - A batch through `awa.insert_jobs_compat_batch` takes all its unique claims before any lane lock, while per-row inserts (generic `insert_many`, single-job inserts, 0.6.14 and earlier binaries) take each job's claim and lane together, lane by lane. A per-row transaction that already holds a lane the batch needs and then inserts a job whose unique key the batch has claimed but not committed now fails with `deadlock detected` (SQLSTATE `40P01`) instead of skipping the job or failing with a unique violation (`23505`). Retry the aborted transaction. Batches through the function do not deadlock with each other this way. ### Added diff --git a/awa-model/migrations/v023_install_queue_storage_substrate.sql b/awa-model/migrations/v023_install_queue_storage_substrate.sql index 3f0957937..6a688758b 100644 --- a/awa-model/migrations/v023_install_queue_storage_substrate.sql +++ b/awa-model/migrations/v023_install_queue_storage_substrate.sql @@ -647,7 +647,9 @@ REVOKE EXECUTE ON FUNCTION awa.repair_lane_sequence(TEXT, TEXT) FROM PUBLIC; -- index on a populated partition holds SHARE, which blocks claim inserts, for -- the whole build (minutes per million batches), so the index is only ever -- created on an empty partition: by the installer, or by claim prune right --- after it truncates a slot. +-- after it truncates a slot. A different relation that already has the +-- index's name is never dropped: the function warns, naming it, and the slot +-- keeps scanning until an operator removes or renames it. -- -- fastupdate is off: job ids arrive in ascending order, so direct inserts -- append to the right edge of the entry tree and cost no more than the @@ -671,6 +673,7 @@ DECLARE v_function REGPROCEDURE; v_function_owner OID; v_grantee TEXT; + v_previous_role TEXT; v_set_mode TEXT := CASE WHEN current_setting('server_version_num')::int >= 160000 THEN 'SET' ELSE 'MEMBER' END; BEGIN @@ -696,6 +699,7 @@ BEGIN LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog, pg_temp + SET quote_all_identifiers = off AS $func$ DECLARE v_child TEXT; @@ -721,19 +725,44 @@ BEGIN USING ERRCODE = '22023'; END IF; + -- Only a valid, non-partial GIN index on job_ids alone, with the + -- default array_ops opclass, serves the by-job lookups. IF EXISTS ( SELECT 1 FROM pg_index AS idx JOIN pg_class AS index_class ON index_class.oid = idx.indexrelid JOIN pg_am AS method ON method.oid = index_class.relam + JOIN pg_attribute AS job_ids + ON job_ids.attrelid = idx.indrelid + AND job_ids.attname = 'job_ids' + AND NOT job_ids.attisdropped + JOIN pg_opclass AS opclass ON opclass.oid = idx.indclass[0] WHERE idx.indrelid = to_regclass(format('%%I.%%I', %1$L, v_child)) AND method.amname = 'gin' AND idx.indisvalid - AND pg_get_indexdef(idx.indexrelid) LIKE '%%USING gin (job_ids)%%' + AND idx.indpred IS NULL + AND idx.indexprs IS NULL + AND idx.indnatts = 1 + AND idx.indkey[0] = job_ids.attnum + AND opclass.opcname = 'array_ops' + AND opclass.opcnamespace = 'pg_catalog'::regnamespace ) THEN RETURN TRUE; END IF; + -- Something else already has the index's name: a partial or + -- otherwise different index, or another relation. Leave it alone + -- and keep scanning; the operator decides what to do with it. + IF to_regclass(format('%%I.%%I', %1$L, v_index)) IS NOT NULL THEN + RAISE WARNING '%%.%% is not a GIN index on job_ids, so claim slot %% of schema %% keeps scanning for job ids', + quote_ident(%1$L), quote_ident(v_index), p_slot, %1$L + USING HINT = format( + 'Drop or rename %%s.%%s; claim prune builds the job_ids index the next time it empties claim slot %%s.', + quote_ident(%1$L), quote_ident(v_index), p_slot + ); + RETURN FALSE; + END IF; + -- NOWAIT: a partition someone is writing to is not empty for long -- enough to index, so give up rather than queue behind claims. -- Leaving the block through the exception rolls back its @@ -747,11 +776,11 @@ BEGIN RAISE EXCEPTION USING ERRCODE = 'object_not_in_prerequisite_state'; END IF; EXECUTE format( - 'CREATE INDEX IF NOT EXISTS %%I ON %%I.%%I USING GIN (job_ids) WITH (fastupdate = off)', + 'CREATE INDEX %%I ON %%I.%%I USING GIN (job_ids) WITH (fastupdate = off)', v_index, %1$L, v_child ); EXCEPTION - WHEN lock_not_available OR object_not_in_prerequisite_state THEN + WHEN lock_not_available OR object_not_in_prerequisite_state OR duplicate_table THEN RETURN FALSE; END; RETURN TRUE; @@ -776,6 +805,7 @@ BEGIN v_function, pg_get_userbyid(v_table_owner) ); + v_function_owner := v_table_owner; ELSE RAISE WARNING '% is owned by %, not the owner % of %.lease_claim_batches, so it cannot create the job_ids index', v_function, pg_get_userbyid(v_function_owner), pg_get_userbyid(v_table_owner), p_schema @@ -787,6 +817,14 @@ BEGIN END IF; END IF; + -- Reset EXECUTE as the function's owner, after any transfer, so a former + -- owner that holds TRUNCATE keeps it. A role that may only SET ROLE to the + -- new owner cannot grant or revoke as itself, so it acts as the owner for + -- the reset and then restores its previous role. + IF NOT pg_has_role(current_user, v_function_owner, 'USAGE') THEN + v_previous_role := current_setting('role'); + PERFORM set_config('role', pg_get_userbyid(v_function_owner), true); + END IF; FOR v_grantee IN SELECT DISTINCT CASE WHEN acl.grantee = 0 THEN 'PUBLIC' ELSE quote_ident(pg_get_userbyid(acl.grantee)) END @@ -806,6 +844,9 @@ BEGIN LOOP EXECUTE format('GRANT EXECUTE ON FUNCTION %s TO %s', v_function, v_grantee); END LOOP; + IF v_previous_role IS NOT NULL THEN + PERFORM set_config('role', v_previous_role, true); + END IF; END; $refresh$; diff --git a/awa-model/src/insert.rs b/awa-model/src/insert.rs index e8e38634e..bea14b936 100644 --- a/awa-model/src/insert.rs +++ b/awa-model/src/insert.rs @@ -21,8 +21,9 @@ use sqlx::{PgExecutor, PgPool}; // `awa.insert_jobs_compat_batch` (the `insert_jobs_compat_batch` schema // patch), which does the lane-independent work first (job ids, unique claims, // payloads) and then registers and reserves its lanes one at a time in -// canonical order, so a lane is held only across its registration, -// reservation, the ready-row writes and the commit. +// canonical order. Each lane's lock is taken when the batch reaches that +// lane rather than at the batch's first ready row, and like every enqueue +// lock it is held until the transaction ends. // // Generic `insert_many` always uses the per-row path, as does // `insert_many_copy` on a schema without the patch: rows are sorted by the diff --git a/awa-model/src/migrations.rs b/awa-model/src/migrations.rs index da16aac78..a4847b498 100644 --- a/awa-model/src/migrations.rs +++ b/awa-model/src/migrations.rs @@ -356,13 +356,17 @@ macro_rules! definer_lane_provisioning_pending_sql { /// Rows naming what still lacks the `claim_batch_job_index` patch: the /// function that installs the per-schema index function, the installer, and /// each queue-storage schema with partitioned `lease_claim_batches` but no -/// SECURITY DEFINER `ensure_claim_batch_job_index`. The indexes do not keep -/// the patch pending: the patch builds none, and claim prune adds each one as -/// it empties a partition. +/// SECURITY DEFINER `ensure_claim_batch_job_index`. A helper or index function +/// from before the index check required a non-partial index (0.6.15) is +/// pending too. The indexes do not keep the patch pending: the patch builds +/// none, and claim prune adds each one as it empties a partition. macro_rules! claim_batch_job_index_pending_sql { () => { "SELECT 'awa.refresh_claim_batch_job_index_helper'::text AS target \ - WHERE to_regprocedure('awa.refresh_claim_batch_job_index_helper(text)') IS NULL \ + WHERE NOT EXISTS ( \ + SELECT 1 FROM pg_proc AS p \ + WHERE p.oid = to_regprocedure('awa.refresh_claim_batch_job_index_helper(text)') \ + AND p.prosrc LIKE '%indpred IS NULL%') \ UNION ALL \ SELECT 'awa.install_queue_storage_substrate' \ WHERE EXISTS ( \ @@ -387,7 +391,9 @@ macro_rules! claim_batch_job_index_pending_sql { WHERE p.pronamespace = n.oid \ AND p.proname = 'ensure_claim_batch_job_index' \ AND p.prosecdef \ - AND oidvectortypes(p.proargtypes) = 'integer')" + AND oidvectortypes(p.proargtypes) = 'integer' \ + AND p.prosrc LIKE '%indpred IS NULL%' \ + AND 'quote_all_identifiers=off' = ANY (p.proconfig))" }; } diff --git a/awa/tests/enqueue_lane_contention_test.rs b/awa/tests/enqueue_lane_contention_test.rs index e411364c8..9276fae9f 100644 --- a/awa/tests/enqueue_lane_contention_test.rs +++ b/awa/tests/enqueue_lane_contention_test.rs @@ -122,7 +122,14 @@ struct Samples { /// transaction. statement_ms: Vec, commit_ms: Vec, + /// Jobs the producers asked to insert in the measured window. + requested: u64, + /// Jobs the inserts returned in the measured window: requested jobs less + /// those skipped for a unique conflict. jobs: u64, + /// Jobs the inserts returned over the whole run, warmup and the batches + /// still in flight at the end of the window included. + acknowledged_total: u64, batches: u64, duplicate_batches: u64, fallback_rows: u64, @@ -134,7 +141,9 @@ impl Samples { self.fallback_row_ms.extend(other.fallback_row_ms); self.statement_ms.extend(other.statement_ms); self.commit_ms.extend(other.commit_ms); + self.requested += other.requested; self.jobs += other.jobs; + self.acknowledged_total += other.acknowledged_total; self.batches += other.batches; self.duplicate_batches += other.duplicate_batches; self.fallback_rows += other.fallback_rows; @@ -286,7 +295,10 @@ const LEGACY_ROW_FUNCTION_SQL: &str = r#" ORDER BY input.ord "#; -async fn legacy_row_function_insert(pool: &PgPool, jobs: &[InsertParams]) -> Result<(), AwaError> { +async fn legacy_row_function_insert( + pool: &PgPool, + jobs: &[InsertParams], +) -> Result { let rows: Vec<_> = jobs .iter() .map(|job| { @@ -294,7 +306,7 @@ async fn legacy_row_function_insert(pool: &PgPool, jobs: &[InsertParams]) -> Res .expect("prepare row") }) .collect(); - sqlx::query(LEGACY_ROW_FUNCTION_SQL) + let inserted = sqlx::query(LEGACY_ROW_FUNCTION_SQL) .bind(rows.iter().map(|row| row.kind()).collect::>()) .bind(rows.iter().map(|row| row.queue()).collect::>()) .bind(rows.iter().map(|row| row.args()).collect::>()) @@ -324,7 +336,7 @@ async fn legacy_row_function_insert(pool: &PgPool, jobs: &[InsertParams]) -> Res ) .fetch_all(pool) .await?; - Ok(()) + Ok(inserted.len()) } #[allow(clippy::too_many_arguments)] @@ -367,57 +379,65 @@ async fn producer_loop( let mut commit_ms = None; let mut fallback_rows_ms: Vec = Vec::new(); - let outcome: Result<(), AwaError> = match scenario.mode { + // Rows the insert returned (or, for the queue-storage batch, reports): + // a skipping insert omits the jobs whose unique key is taken. + let outcome: Result = match scenario.mode { Mode::InsertMany if scenario.split_commit => { let mut tx = pool.begin().await.expect("begin"); let statement_started = Instant::now(); let result = insert_many(&mut *tx, &jobs).await; let statement_elapsed = statement_started.elapsed(); match result { - Ok(_) => { + Ok(rows) => { let commit_started = Instant::now(); tx.commit().await.expect("commit"); statement_ms = Some(statement_elapsed.as_secs_f64() * 1e3); commit_ms = Some(commit_started.elapsed().as_secs_f64() * 1e3); - Ok(()) + Ok(rows.len()) } Err(err) => Err(err), } } - Mode::InsertMany => insert_many(&pool, &jobs).await.map(|_| ()), - Mode::InsertManyCopy => insert_many_copy_from_pool(&pool, &jobs).await.map(|_| ()), + Mode::InsertMany => insert_many(&pool, &jobs).await.map(|rows| rows.len()), + Mode::InsertManyCopy => insert_many_copy_from_pool(&pool, &jobs) + .await + .map(|rows| rows.len()), Mode::InsertManySkip => insert_many_skip_unique_conflicts(&pool, &jobs) .await - .map(|_| ()), - Mode::QueueStorage => store.enqueue_params_batch(&pool, &jobs).await.map(|_| ()), + .map(|rows| rows.len()), + Mode::QueueStorage => store.enqueue_params_batch(&pool, &jobs).await, Mode::LegacyRowFunction => legacy_row_function_insert(&pool, &jobs).await, }; let mut fallback_count = 0u64; - match outcome { - Ok(()) => {} + let inserted: u64 = match outcome { + Ok(rows) => rows as u64, Err(err) if is_unique_violation(&err) => { // What a replaying producer does on a unique conflict: // insert the batch again one row at a time, skipping // duplicates. + let mut fallback_inserted = 0u64; for job in &jobs { let row_started = Instant::now(); match insert_many(&pool, std::slice::from_ref(job)).await { - Ok(_) => {} + Ok(rows) => fallback_inserted += rows.len() as u64, Err(err) if is_unique_violation(&err) => {} Err(err) => panic!("fallback insert failed: {err}"), } fallback_rows_ms.push(row_started.elapsed().as_secs_f64() * 1e3); fallback_count += 1; } + fallback_inserted } Err(err) => panic!("enqueue failed: {err}"), - } + }; let elapsed = started.elapsed(); + samples.acknowledged_total += inserted; if started >= measure_from { samples.batch_ms.push(elapsed.as_secs_f64() * 1e3); - samples.jobs += scenario.batch_size as u64; + samples.requested += scenario.batch_size as u64; + samples.jobs += inserted; samples.batches += 1; if let Some(ms) = statement_ms { samples.statement_ms.push(ms); @@ -542,8 +562,8 @@ fn print_result(result: &ScenarioResult) { println!( "[lane-bench] {name} mode={mode} producers={producers} batch={batch} unique={unique} \ shards={shards} dup_pct={dup} | {rate:.0} jobs/s | batch ms p50={p50:.1} p95={p95:.1} \ - p99={p99:.1} max={max:.1} | waiters_mean={waiters:.1} | landed={landed} \ - distinct_seqs={distinct}", + p99={p99:.1} max={max:.1} | waiters_mean={waiters:.1} | inserted={inserted} of \ + requested={requested} | landed={landed} distinct_seqs={distinct}", name = scenario.name, mode = scenario.mode.label(), producers = scenario.producers, @@ -557,6 +577,8 @@ fn print_result(result: &ScenarioResult) { p99 = result.p99_ms, max = result.max_ms, waiters = result.mean_waiters, + inserted = result.samples.jobs, + requested = result.samples.requested, landed = result.landed, distinct = result.distinct_lane_seqs, ); @@ -595,6 +617,7 @@ fn print_result(result: &ScenarioResult) { fn emit_json(result: &ScenarioResult) { let mut outcomes = HashMap::new(); outcomes.insert("jobs".to_string(), result.samples.jobs); + outcomes.insert("requested".to_string(), result.samples.requested); outcomes.insert("batches".to_string(), result.samples.batches); BenchmarkResult { schema_version: SCHEMA_VERSION, @@ -624,9 +647,9 @@ fn emit_json(result: &ScenarioResult) { .emit(); } -/// Lane integrity: every job that was acknowledged landed exactly once, under -/// a distinct `(shard, lane_seq)`, and every shard's sequence numbers are -/// dense (no producer lost or duplicated a reservation). +/// Lane integrity: exactly the jobs the inserts returned landed, each once, +/// under a distinct `(shard, lane_seq)` (no producer lost or duplicated a +/// reservation). fn assert_lane_integrity(result: &ScenarioResult) { if let Some((claimed, _)) = result.drained { assert_eq!( @@ -640,12 +663,10 @@ fn assert_lane_integrity(result: &ScenarioResult) { "{}: ready rows share a (shard, lane_seq)", result.scenario.name ); - assert!( - result.landed as u64 >= result.samples.jobs, - "{}: {} jobs acknowledged in the window but only {} rows landed", - result.scenario.name, - result.samples.jobs, - result.landed + assert_eq!( + result.landed as u64, result.samples.acknowledged_total, + "{}: rows landed against jobs the inserts returned", + result.scenario.name ); } diff --git a/awa/tests/insert_many_batch_test.rs b/awa/tests/insert_many_batch_test.rs index 63c36b676..b7446cd12 100644 --- a/awa/tests/insert_many_batch_test.rs +++ b/awa/tests/insert_many_batch_test.rs @@ -2,14 +2,15 @@ //! `insert_jobs_compat_batch` schema patch. //! //! A batch does its lane-independent work (job ids, unique claims) before it -//! registers and reserves its lanes one at a time in canonical order, and -//! holds a lane lock only for the registration, the reservation, the ready-row -//! writes and the commit. These tests cover that ordering, skip-on-conflict +//! registers and reserves its lanes one at a time in canonical order, taking +//! each lane's lock only when it reaches that lane; the lock is then held until +//! the transaction ends. These tests cover that ordering, skip-on-conflict //! inserts, and the shape of what lands in the lanes. //! //! Requires a running Postgres instance. //! Run with: `cargo test --package awa --test insert_many_batch_test -- --nocapture` +use awa::model::unique::compute_unique_key; use awa::model::{ insert_many, insert_many_copy_from_pool, insert_many_skip_unique_conflicts, AwaError, QueueStorage, @@ -376,6 +377,154 @@ async fn canonical_storage_inserts_and_skips_row_by_row() { ); } +/// Call `awa.insert_jobs_compat_batch` directly with `p_skip_unique_conflicts` +/// set explicitly. The Rust entry points pass `true` whenever a batch has a +/// unique key, so only a direct caller reaches the failing mode with one. +async fn insert_through_batch_function( + pool: &PgPool, + jobs: &[InsertParams], + skip_unique_conflicts: bool, +) -> Result, sqlx::Error> { + let unique_keys: Vec>> = jobs + .iter() + .map(|job| { + job.opts.unique.as_ref().map(|unique| { + compute_unique_key( + &job.kind, + unique.by_queue.then_some(job.opts.queue.as_str()), + unique.by_args.then_some(&job.args), + unique.by_period, + ) + }) + }) + .collect(); + // PostgreSQL's leftmost bit is UniqueOpts bit 0. + let unique_states: Vec> = jobs + .iter() + .map(|job| { + job.opts.unique.as_ref().map(|unique| { + (0..8) + .map(|bit| { + if unique.states & (1 << bit) != 0 { + '1' + } else { + '0' + } + }) + .collect() + }) + }) + .collect(); + sqlx::query_as( + r#" + SELECT * + FROM awa.insert_jobs_compat_batch( + $1::text[], $2::text[], $3::jsonb[], + array_fill('available'::text, ARRAY[cardinality($1::text[])]), + array_fill(2::smallint, ARRAY[cardinality($1::text[])]), + array_fill(25::smallint, ARRAY[cardinality($1::text[])]), + array_fill(NULL::timestamptz, ARRAY[cardinality($1::text[])]), + array_fill('{}'::jsonb, ARRAY[cardinality($1::text[])]), + array_fill('[]'::jsonb, ARRAY[cardinality($1::text[])]), + $4::bytea[], $5::text[], + array_fill(NULL::text, ARRAY[cardinality($1::text[])]), + $6 + ) + "#, + ) + .bind(jobs.iter().map(|job| job.kind.as_str()).collect::>()) + .bind( + jobs.iter() + .map(|job| job.opts.queue.as_str()) + .collect::>(), + ) + .bind(jobs.iter().map(|job| &job.args).collect::>()) + .bind(unique_keys) + .bind(unique_states) + .bind(skip_unique_conflicts) + .fetch_all(pool) + .await +} + +async fn jobs_in_queue(pool: &PgPool, queue: &str) -> Vec { + sqlx::query_scalar("SELECT (args->>'seq')::bigint FROM awa.jobs WHERE queue = $1 ORDER BY 1") + .bind(queue) + .fetch_all(pool) + .await + .expect("read queue jobs") +} + +/// With `p_skip_unique_conflicts = false` a unique conflict, with an already +/// claimed key or within the batch, fails the whole batch with `23505` and +/// leaves nothing behind, on queue storage and on canonical storage. +#[tokio::test] +async fn batch_function_without_skip_fails_the_batch_on_a_unique_conflict() { + let _guard = ENGINE_GUARD.lock().await; + for queue_storage in [true, false] { + let pool = if queue_storage { + queue_storage_pool().await + } else { + let pool = setup::setup(8).await; + setup::reset_to_canonical(&pool).await; + pool + }; + let queue = unique_queue(if queue_storage { + "noskip_qs" + } else { + "noskip_canonical" + }); + insert_many(&pool, &[job(&queue, 1, true)]) + .await + .expect("seed"); + + for batch in [ + vec![ + job(&queue, 0, true), + job(&queue, 1, true), + job(&queue, 2, true), + ], + vec![ + job(&queue, 3, true), + job(&queue, 4, true), + job(&queue, 4, true), + ], + ] { + let err = insert_through_batch_function(&pool, &batch, false) + .await + .expect_err("a unique conflict must fail the batch"); + assert!( + is_unique_violation(&AwaError::Database(err)), + "queue_storage={queue_storage}" + ); + assert_eq!( + jobs_in_queue(&pool, &queue).await, + vec![1], + "queue_storage={queue_storage}: nothing from the failed batch landed" + ); + } + + let inserted = insert_through_batch_function( + &pool, + &[ + job(&queue, 0, true), + job(&queue, 1, true), + job(&queue, 2, true), + ], + true, + ) + .await + .expect("the same batch with skipping"); + assert_eq!( + inserted + .iter() + .map(|row| row.args["seq"].as_i64().unwrap()) + .collect::>(), + vec![0, 2], + "queue_storage={queue_storage}" + ); + } +} + /// While another transaction holds a lane lock, a batch bound for that lane /// has already done its lane-independent work (it holds its unique claims) and /// has not yet written a ready row. Every row's unique claim, not just the diff --git a/awa/tests/lane_ownership_roles_test.rs b/awa/tests/lane_ownership_roles_test.rs index f7b85eea0..99a2ec5a8 100644 --- a/awa/tests/lane_ownership_roles_test.rs +++ b/awa/tests/lane_ownership_roles_test.rs @@ -893,6 +893,168 @@ async fn set_only_schema_owner_membership_repairs_legacy_lanes_and_restores_role cluster.teardown().await } +/// Who may execute the default schema's `ensure_claim_batch_job_index`. +#[derive(Debug, PartialEq, sqlx::FromRow)] +struct ClaimBatchIndexAcl { + owner: String, + public: bool, + runtime: bool, + migrator: bool, +} + +async fn claim_batch_job_index_function_acl( + cluster: &Cluster, +) -> Result { + sqlx::query_as( + "SELECT pg_get_userbyid(routine.proowner)::text AS owner, \ + has_function_privilege('public', routine.oid, 'EXECUTE') AS public, \ + has_function_privilege($1, routine.oid, 'EXECUTE') AS runtime, \ + has_function_privilege($2, routine.oid, 'EXECUTE') AS migrator \ + FROM pg_proc AS routine \ + WHERE routine.oid = to_regprocedure('awa.ensure_claim_batch_job_index(integer)')", + ) + .bind(&cluster.runtime_name) + .bind(&cluster.migrator_name) + .fetch_one(&cluster.owner) + .await +} + +/// Give `ensure_claim_batch_job_index` to the migration role, executable by +/// `PUBLIC`, as a refresh by the migration role acting as itself would +/// create it. Needs the migration role to hold the owner's privileges. +async fn hand_claim_batch_job_index_to_migrator(cluster: &Cluster) -> TestResult { + cluster + .migrator + .execute(sqlx::raw_sql(&sql(format!( + "ALTER FUNCTION awa.ensure_claim_batch_job_index(integer) OWNER TO {migrator}; \ + GRANT EXECUTE ON FUNCTION awa.ensure_claim_batch_job_index(integer) TO PUBLIC", + migrator = cluster.migrator_name + )))) + .await?; + Ok(()) +} + +/// `ensure_claim_batch_job_index` ends up owned by the table owner and +/// executable by exactly the roles holding `TRUNCATE` on +/// `lease_claim_batches`, never `PUBLIC`, however the migration role refreshes +/// it: as itself with ordinary membership in the owner, as itself with +/// SET-only membership (PostgreSQL 16+: it may `SET ROLE` to the owner but +/// does not inherit its privileges), and through the `claim_batch_job_index` +/// patch with the owner as effective role, as the database-roles guide says. +/// The migration role holds `TRUNCATE` itself, so it keeps `EXECUTE` after +/// handing the function over, even when it no longer inherits the owner's. +#[tokio::test] +async fn set_only_schema_owner_membership_resets_claim_batch_job_index_execute() -> TestResult { + let cluster = Cluster::new(Schema::Current, false).await?; + let server_version: i32 = + sqlx::query_scalar("SELECT current_setting('server_version_num')::int") + .fetch_one(&cluster.owner) + .await?; + if server_version < 160_000 { + return cluster.teardown().await; + } + let expected = ClaimBatchIndexAcl { + owner: cluster.owner_name.clone(), + public: false, + runtime: true, + migrator: true, + }; + assert_eq!( + claim_batch_job_index_function_acl(&cluster).await?, + expected + ); + + cluster + .owner + .execute(sqlx::raw_sql(&sql(format!( + "GRANT USAGE, CREATE ON SCHEMA awa TO {migrator}; \ + GRANT TRUNCATE ON awa.lease_claim_batches TO {migrator}; \ + GRANT EXECUTE ON FUNCTION awa.refresh_claim_batch_job_index_helper(text) TO {migrator}", + migrator = cluster.migrator_name + )))) + .await?; + let refresh = "SELECT awa.refresh_claim_batch_job_index_helper('awa')"; + + // Ordinary membership: the migration role holds the owner's privileges. + hand_claim_batch_job_index_to_migrator(&cluster).await?; + sqlx::query(refresh).execute(&cluster.migrator).await?; + assert_eq!( + claim_batch_job_index_function_acl(&cluster).await?, + expected, + "a refresh by a migration role that inherits the owner" + ); + + // SET-only membership. + hand_claim_batch_job_index_to_migrator(&cluster).await?; + cluster + .admin + .execute(sqlx::raw_sql(&sql(format!( + "GRANT {owner} TO {migrator} WITH INHERIT FALSE; \ + GRANT {owner} TO {migrator} WITH SET TRUE", + owner = cluster.owner_name, + migrator = cluster.migrator_name + )))) + .await?; + let mut conn = cluster.migrator.acquire().await?; + let (inherits_owner, can_set_owner): (bool, bool) = sqlx::query_as( + "SELECT pg_has_role(current_user, $1, 'USAGE'), pg_has_role(current_user, $1, 'SET')", + ) + .bind(&cluster.owner_name) + .fetch_one(&mut *conn) + .await?; + assert!(!inherits_owner); + assert!(can_set_owner); + let role_query = "SELECT current_user::text, current_setting('role')"; + let role_before: (String, String) = sqlx::query_as(role_query).fetch_one(&mut *conn).await?; + sqlx::query(refresh).execute(&mut *conn).await?; + let role_after: (String, String) = sqlx::query_as(role_query).fetch_one(&mut *conn).await?; + assert_eq!( + role_after, role_before, + "the refresh restores the caller's role" + ); + drop(conn); + assert_eq!( + claim_batch_job_index_function_acl(&cluster).await?, + expected, + "a refresh by a migration role that may only SET ROLE to the owner" + ); + + // The whole patch, with the owner as the effective role. + cluster + .owner + .execute("DROP FUNCTION awa.ensure_claim_batch_job_index(integer)") + .await?; + let migrate_pool = cluster.migrator_as_owner().await?; + assert_eq!( + migrations::pending_schema_patches(&migrate_pool) + .await? + .iter() + .map(|patch| patch.name) + .collect::>(), + ["claim_batch_job_index"] + ); + let report = migrations::run_with_report(&migrate_pool).await?; + let role_after: (String, String) = sqlx::query_as(role_query).fetch_one(&migrate_pool).await?; + migrate_pool.close().await; + assert_eq!( + role_after, + (cluster.owner_name.clone(), cluster.owner_name.clone()) + ); + assert!(report.complete(), "{:?}", report.patches); + let patch = report + .patches + .iter() + .find(|patch| patch.name == "claim_batch_job_index") + .expect("claim_batch_job_index is a schema patch"); + assert_eq!(patch.outcome, migrations::PatchOutcome::Applied); + assert_eq!( + claim_batch_job_index_function_acl(&cluster).await?, + expected, + "the patch applied by the migration role as the owner" + ); + cluster.teardown().await +} + /// Reassigning a lane sequence waits on any open enqueue or claim on the lane, /// so the repair bounds that wait and holds each sequence's lock only for its /// own short transaction. A held sequence is skipped and reported; the rest diff --git a/awa/tests/patch_upgrade_rehearsal_test.rs b/awa/tests/patch_upgrade_rehearsal_test.rs index be55646d7..d64df92a8 100644 --- a/awa/tests/patch_upgrade_rehearsal_test.rs +++ b/awa/tests/patch_upgrade_rehearsal_test.rs @@ -754,8 +754,8 @@ async fn released_patch_upgrade_and_rollback() { .unwrap(); assert_eq!( pending.iter().map(|patch| patch.name).collect::>(), - ["prune_controls"], - "a released 0.6.15 deployment needs only prune controls" + ["claim_batch_job_index", "prune_controls"], + "a released 0.6.15 deployment needs the claim batch index fix and prune controls" ); } @@ -824,9 +824,12 @@ async fn released_patch_upgrade_and_rollback() { if released_version == "0.6.9" { assert!(patch_sql.contains("R__prepared_lane_sequences")); } else { - assert_eq!(patch_sql.matches("-- Schema patch R__").count(), 1); + assert_eq!(patch_sql.matches("-- Schema patch R__").count(), 2); assert!(!patch_sql.contains("R__prepared_lane_sequences")); - assert!(!patch_sql.contains("R__claim_batch_job_index")); + assert!( + patch_sql.contains("R__claim_batch_job_index"), + "0.6.15's claim batch index function is replaced" + ); assert!(!patch_sql.contains("R__insert_jobs_compat_batch")); } assert!( @@ -971,6 +974,7 @@ async fn released_patch_upgrade_and_rollback() { "already present" } ), + "Schema patch claim_batch_job_index: applied".to_string(), "Schema patch prune_controls: applied".to_string(), format!("Storage routing: unchanged (state active, engine queue_storage, schema {SCHEMA}"), ] { diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index d9172f145..c1b712520 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -21263,7 +21263,13 @@ async fn test_claim_batch_job_index_is_built_only_on_empty_partitions() { .await .expect("Failed to read ensure_claim_batch_job_index"); assert!(definer); - assert_eq!(config, vec!["search_path=pg_catalog, pg_temp".to_string()]); + assert_eq!( + config, + vec![ + "search_path=pg_catalog, pg_temp".to_string(), + "quote_all_identifiers=off".to_string(), + ] + ); assert!(!public_execute, "PUBLIC must not execute {function}"); let grantee = "awa_qs_cbji_truncate_holder"; @@ -21407,6 +21413,117 @@ async fn test_claim_batch_job_index_is_built_only_on_empty_partitions() { assert_eq!(claim_batch_job_index_slots(&pool, schema).await, vec![0, 1]); } +/// `(index name, partial)` for every GIN index on one claim partition. +async fn claim_partition_gin_indexes( + pool: &sqlx::PgPool, + schema: &str, + slot: i32, +) -> Vec<(String, bool)> { + sqlx::query_as( + r#" + SELECT index_class.relname::text, index_meta.indpred IS NOT NULL + FROM pg_index AS index_meta + JOIN pg_class AS index_class ON index_class.oid = index_meta.indexrelid + JOIN pg_am AS method ON method.oid = index_class.relam + WHERE index_meta.indrelid = to_regclass(format('%I.lease_claim_batches_%s', $1::text, $2::int)) + AND method.amname = 'gin' + ORDER BY 1 + "#, + ) + .bind(schema) + .bind(slot) + .fetch_all(pool) + .await + .expect("Failed to list claim partition GIN indexes") +} + +/// Only a non-partial GIN index on `job_ids` counts as the claim batch index. +/// A partial one under another name does not stop the real index being +/// built; an index that already has the real index's name but another +/// definition is reported and left in place, never dropped or mistaken for +/// the real one. The check does not depend on the caller's +/// `quote_all_identifiers`. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn test_claim_batch_job_index_requires_a_full_job_ids_index() { + let (_db_guard, pool) = setup_pool(4).await; + let schema = "awa_qs_cbji_shape"; + let _store = create_store_with_config( + &pool, + QueueStorageConfig { + schema: schema.to_string(), + queue_slot_count: 4, + lease_slot_count: 2, + queue_stripe_count: 1, + lease_claim_receipts: true, + claim_slot_count: 2, + }, + ) + .await; + drop_claim_batch_job_index(&pool, schema, 0).await; + drop_claim_batch_job_index(&pool, schema, 1).await; + + sqlx::raw_sql(&format!( + "CREATE INDEX lease_claim_batches_0_job_ids_idx \ + ON {schema}.lease_claim_batches_0 USING GIN (job_ids) \ + WHERE cardinality(job_ids) > 1; \ + CREATE INDEX operator_partial_job_ids \ + ON {schema}.lease_claim_batches_1 USING GIN (job_ids) \ + WHERE cardinality(job_ids) > 1" + )) + .execute(&pool) + .await + .expect("create partial job_ids indexes"); + + assert!( + !ensure_claim_batch_job_index(&pool, schema, 0).await, + "a partial index under the real index's name is not the real index" + ); + assert_eq!( + claim_partition_gin_indexes(&pool, schema, 0).await, + vec![("lease_claim_batches_0_job_ids_idx".to_string(), true)], + "the conflicting index is left in place" + ); + + assert!(ensure_claim_batch_job_index(&pool, schema, 1).await); + assert_eq!( + claim_partition_gin_indexes(&pool, schema, 1).await, + vec![ + ("lease_claim_batches_1_job_ids_idx".to_string(), false), + ("operator_partial_job_ids".to_string(), true), + ], + "a partial index under another name does not stand in for the real one" + ); + + // With the real index present the check returns before it would try + // to lock the partition, so a busy partition still reports indexed, even + // for a caller that quotes every identifier. + let mut writer = pool.begin().await.unwrap(); + sqlx::query(&format!( + "LOCK TABLE {schema}.lease_claim_batches_1 IN ROW EXCLUSIVE MODE" + )) + .execute(&mut *writer) + .await + .unwrap(); + let mut caller = pool.acquire().await.unwrap(); + sqlx::query("SET quote_all_identifiers = on") + .execute(&mut *caller) + .await + .unwrap(); + let indexed = tokio::time::timeout( + ci_timing::scaled_timeout(Duration::from_secs(5)), + ensure_claim_batch_job_index(&mut *caller, schema, 1), + ) + .await + .expect("ensure_claim_batch_job_index must not wait for a busy partition"); + sqlx::query("RESET quote_all_identifiers") + .execute(&mut *caller) + .await + .unwrap(); + drop(caller); + writer.rollback().await.unwrap(); + assert!(indexed, "the existing job_ids index must be recognised"); +} + /// The `claim_batch_job_index` patch is versionless and idempotent, installs /// the index function without building any index, and claim prune then /// indexes each slot it truncates. A schema without the patch keeps working @@ -21583,6 +21700,36 @@ async fn test_claim_batch_job_index_patch_leaves_indexing_to_claim_prune() { .find(|patch| patch.name == "claim_batch_job_index") .unwrap(); assert_eq!(patch.outcome, migrations::PatchOutcome::AlreadyPresent); + + // An index function from before the check pinned its settings (0.6.15) + // leaves the patch pending until it is replaced. + sqlx::raw_sql(&format!( + "ALTER FUNCTION {schema}.ensure_claim_batch_job_index(integer) \ + RESET quote_all_identifiers" + )) + .execute(&pool) + .await + .expect("unpin the index function's settings"); + let status = migrations::schema_patch_status(&pool) + .await + .unwrap() + .into_iter() + .find(|status| status.name == "claim_batch_job_index") + .unwrap(); + assert!(!status.applied); + assert_eq!( + status.needs_repair, + vec![format!("claim batch job index in schema {schema}")] + ); + let report = migrations::run_with_report(&pool).await.unwrap(); + let patch = report + .patches + .iter() + .find(|patch| patch.name == "claim_batch_job_index") + .unwrap(); + assert_eq!(patch.outcome, migrations::PatchOutcome::Applied); + assert!(patch.still_needs_repair.is_empty(), "{patch:?}"); + let patch_sql = migrations::SCHEMA_PATCHES .iter() .find(|patch| patch.name == "claim_batch_job_index") diff --git a/docs/migrations.md b/docs/migrations.md index 2ae638fbf..df2a3951e 100644 --- a/docs/migrations.md +++ b/docs/migrations.md @@ -247,7 +247,7 @@ That writes one `V__.sql` file per migration, plus one `R__.ensure_claim_batch_job_index(slot)` in every queue-storage schema and builds no index itself, so it takes no lock on any awa or queue-storage table and is safe inside a long transaction under live traffic. Claim prune calls that function right after truncating a claim partition, which builds the GIN index on `lease_claim_batches.job_ids` while the partition is empty, so the index covers the ring within one claim-ring cycle. Do not build the index yourself on a populated partition: the build holds a `SHARE` lock, which blocks claim inserts into that partition, for about half a minute per 100k batches. Run the script as the owner of the queue-storage tables. Released 0.6 binaries neither use nor build the index and are unaffected by it, and binaries that use the index read the same rows, by scan, from a schema or partition without it. +`R__claim_batch_job_index.sql` indexes compact claim batches by job id. It installs `.ensure_claim_batch_job_index(slot)` in every queue-storage schema and builds no index itself, so it takes no lock on any awa or queue-storage table and is safe inside a long transaction under live traffic. Claim prune calls that function right after truncating a claim partition, which builds the GIN index on `lease_claim_batches.job_ids` while the partition is empty, so the index covers the ring within one claim-ring cycle. Only a valid, non-partial GIN index on `job_ids` counts; a different index or relation already named `lease_claim_batches__job_ids_idx` is left in place with a warning naming it, and that slot keeps scanning until it is dropped or renamed. Do not build the index yourself on a populated partition: the build holds a `SHARE` lock, which blocks claim inserts into that partition, for about half a minute per 100k batches. Run the script as the owner of the queue-storage tables. Released 0.6 binaries neither use nor build the index and are unaffected by it, and binaries that use the index read the same rows, by scan, from a schema or partition without it. `awa migrate` reports the schema version, each patch and what it updated, and storage routing before and after. It exits 1 if a patch is still missing somewhere, such as a queue-storage schema the migrating role cannot use, and names it. `awa::migrations::schema_patch_status` gives the same per-patch view from Rust. diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index ff76df346..317794450 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -242,7 +242,7 @@ Use an upsert: first-enqueue may create lane rows before any operator inserts a Many connections wait with `wait_event = 'advisory'` while the database is otherwise idle, batch inserts take seconds although the work per batch is milliseconds, and promotion into the same queue runs late while producers saturate it. -On queue storage every enqueue holds a transaction-scoped advisory lock on its `(queue, priority, shard)` lane until it commits, so concurrent producers for one lane run one at a time; a lane's ceiling is a batch's rows divided by the time the lock is held. Generic `insert_many` and per-row COPY take the lock at the batch's first ready row and hold it while the rest of the batch is processed. With the `insert_jobs_compat_batch` schema patch, `insert_many_copy` and `insert_many_skip_unique_conflicts` prepare the whole batch (job ids, unique claims, payloads) before they take any lane lock, and hold it only for the lane write and the commit. +On queue storage every enqueue holds a transaction-scoped advisory lock on its `(queue, priority, shard)` lane until it commits, so concurrent producers for one lane run one at a time; a lane's ceiling is a batch's rows divided by the time the lock is held. Generic `insert_many` and per-row COPY take the lock at the batch's first ready row and hold it while the rest of the batch is processed. With the `insert_jobs_compat_batch` schema patch, `insert_many_copy` and `insert_many_skip_unique_conflicts` prepare the whole batch (job ids, unique claims, payloads) before they take any lane lock, and take each lane's lock when they write that lane; it is still held until the transaction ends. - Run `awa migrate` and move hot producers to `insert_many_copy` or `insert_many_skip_unique_conflicts`. - When `insert_many` fails on a unique conflict, retry with `insert_many_skip_unique_conflicts` rather than inserting the batch again row by row: each single-row insert queues for the lane separately. diff --git a/docs/upgrade-0.6.14-to-0.6.15.md b/docs/upgrade-0.6.14-to-0.6.15.md index 66da7e59f..820eee69c 100644 --- a/docs/upgrade-0.6.14-to-0.6.15.md +++ b/docs/upgrade-0.6.14-to-0.6.15.md @@ -34,7 +34,7 @@ The function is `SECURITY DEFINER` and owned by the owner of `lease_claim_batche The patch adds one function, `awa.insert_jobs_compat_batch`, and changes nothing else: no table DDL, no table or sequence locks, no change to `awa.insert_job_compat` or to storage routing. -The function inserts a whole batch the way `awa.insert_job_compat` inserts one job, with the same validation, unique claims, lane assignment and storage routing. It does everything that does not depend on a lane first (job ids, unique claims in key order, scheduled rows, the result rows). Then, in `(queue, priority, shard)` order, it registers each lane it is the first to use and reserves each lane's range of positions once, as every other producer registers and reserves lane by lane, and writes the ready rows. Each lane's enqueue lock is therefore held only for the reservation, the ready-row write and the commit, instead of from the batch's first ready row to the commit. Concurrent producers and promotion on the same lane wait for that much shorter window. +The function inserts a whole batch the way `awa.insert_job_compat` inserts one job, with the same validation, unique claims, lane assignment and storage routing. It does everything that does not depend on a lane first (job ids, unique claims in key order, scheduled rows, the result rows). Then, in `(queue, priority, shard)` order, it registers each lane it is the first to use and reserves each lane's range of positions once, as every other producer registers and reserves lane by lane, and writes the ready rows. Each lane's enqueue lock is therefore taken when the batch reaches that lane instead of at the batch's first ready row. The lock is transaction-scoped: it is held until the calling transaction commits or rolls back, including any work the caller does after the batch. Concurrent producers and promotion on the same lane no longer wait for the batch's lane-independent work or its earlier lanes. What callers see: diff --git a/docs/upgrade-0.6.15-to-0.6.16.md b/docs/upgrade-0.6.15-to-0.6.16.md index 31c40d89b..5abeacbc0 100644 --- a/docs/upgrade-0.6.15-to-0.6.16.md +++ b/docs/upgrade-0.6.15-to-0.6.16.md @@ -12,7 +12,7 @@ awa --database-url "$AWA_MIGRATOR_DATABASE_URL" migrate ``` - The schema stays at v40. The repeatable `prune_controls` patch runs after `claim_batch_job_index` and `insert_jobs_compat_batch`. + The schema stays at v40. The output reports `Schema patch claim_batch_job_index: applied` and `Schema patch prune_controls: applied`. 3. Apply the [runtime and operator grants](runtime-controls.md#install). Runtime roles need `SELECT` on `awa.runtime_controls` and `SELECT, INSERT, UPDATE, DELETE` on `awa.runtime_control_observations`. Operators need `SELECT` on both tables and `awa.runtime_instances`, plus explicit `EXECUTE` on `awa.halt_runtime_scope(text, text, text)` and `awa.resume_runtime_scope(text, uuid)`. Neither role should have direct control-table writes. Reapply these exceptions after blanket grants and remove inherited mutation privileges from the granting role. 4. Roll workers, producers and admin processes to 0.6.16 after applying the grants. Compatible runtimes discover the patch without a restart or leadership change. They can also run before installation, retaining the existing reclamation policy while catalog observations remain fresh. Once they detect installation, missing read grants close prune admission when any existing permission expires, within five seconds; newly started runtimes have no permission to prune. Other scopes continue. If compatible runtimes are already deployed, expect this pause between migration and grants, and complete the grants before expecting pruning to resume. @@ -22,11 +22,19 @@ The 0.6.15 lifecycle-hook and finalizer bounds remain in place. Ring defaults, s ## What the migration changes +### `claim_batch_job_index` + +0.6.16 replaces the 0.6.15 `.ensure_claim_batch_job_index(slot)`, so `awa migrate` applies this patch again. The 0.6.15 function counted a partial GIN index on `job_ids` as the claim batch index, and also any index already holding the index's name, so claim prune could skip building the real one. The new function accepts only a valid, non-partial GIN index on `job_ids`. When something else already has the name `lease_claim_batches__job_ids_idx`, it raises a warning naming it and leaves it in place; drop or rename it and claim prune builds the index the next time it empties that slot. + +As in 0.6.15 the patch replaces functions only: it builds no index and takes no lock on any `awa` or queue-storage table. Binaries from 0.6.15 call the new function unchanged. + +### `prune_controls` + The patch creates `awa.runtime_controls` and `awa.runtime_control_observations` as ordinary tables, plus functions. The observation table has `instance_id uuid PRIMARY KEY` and `observation jsonb NOT NULL`, with no foreign key. It does not alter `awa.runtime_instances` or any other existing table, refresh the `v023` substrate installer or claim functions, or change storage routing. The observation JSON includes protocol version `1`, report sequence and server `received_at`; publishing keeps the latest sequence for each instance. Re-running the patch preserves the current halt and explicit operator grants. An interrupted migration can be run again. -For an external migration runner, apply `R__prune_controls.sql` after `V40` and the existing `R__claim_batch_job_index.sql` and `R__insert_jobs_compat_batch.sql` patches. Extract the full repeatable set with: +For an external migration runner, apply the new `R__claim_batch_job_index.sql` again, then `R__prune_controls.sql`, after `V40` and the existing `R__insert_jobs_compat_batch.sql` patch. Extract the full repeatable set with: ```bash awa --database-url "$AWA_MIGRATOR_DATABASE_URL" migrate --extract-to ./sql/awa --from 40 @@ -61,4 +69,4 @@ A `halted` status is not proof that PostgreSQL released every lock or that stale ## Roll back -Before rolling back to 0.6.15, fence processes that could prune or explicitly accept losing the halt guarantee: 0.6.15 ignores the control, even when a halt remains stored. Leave the patch installed; its new tables and functions are safe for older binaries, and replay does not resume the fleet. +Before rolling back to 0.6.15, fence processes that could prune or explicitly accept losing the halt guarantee: 0.6.15 ignores the control, even when a halt remains stored. Leave the patches installed; the `prune_controls` tables and functions are safe for older binaries, and replay does not resume the fleet. 0.6.15 binaries call the new `ensure_claim_batch_job_index` unchanged, and a 0.6.15 `awa migrate` leaves it in place.