From 3f1953f7eaf987042f46b7a8556a6ad97da825b2 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Mon, 9 Nov 2026 13:02:31 +1300 Subject: [PATCH 1/6] fix(0.6): require a full job_ids GIN index before skipping the claim batch index build `ensure_claim_batch_job_index` decided a claim partition was already indexed by matching `USING gin (job_ids)` in the index definition text. A partial GIN index on `job_ids` matched, so claim prune never built the real one, and with `quote_all_identifiers` on the text did not match at all. An index of any definition already holding the target name made `CREATE INDEX IF NOT EXISTS` a silent no-op that still reported the slot indexed. The check now reads `pg_index` directly: a valid GIN index with no predicate and no expressions on `job_ids` alone, with the default `array_ops` opclass. When something else already has the target name, the function raises a warning naming it, returns false and leaves it in place; the slot keeps scanning until an operator drops or renames it. The index is created without `IF NOT EXISTS`, so a name taken concurrently also reports false. The generated function pins `quote_all_identifiers = off`. The patch probe now requires the new helper and index function, so a 0.6.15 database sees `claim_batch_job_index` as pending and `awa migrate` replaces the function. The patch still replaces functions only and takes no relation locks. --- CHANGELOG.md | 5 + .../v023_install_queue_storage_substrate.sql | 36 ++++- awa-model/src/migrations.rs | 16 +- awa/tests/patch_upgrade_rehearsal_test.rs | 12 +- awa/tests/queue_storage_runtime_test.rs | 149 +++++++++++++++++- docs/migrations.md | 2 +- docs/upgrade-0.6.15-to-0.6.16.md | 14 +- 7 files changed, 216 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b811375c9..ee8834e95 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,11 +7,16 @@ 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 - Durable operator halt/resume for `maintenance.prune`, in drain mode. Compatible leaders observe it on their existing session; dispatch, cron, promotion, rescue and enqueue continue. Status lists live runtimes without supported, recent reports and cannot report the fleet halted while any remain. Older binaries ignore the control. See [runtime controls](docs/runtime-controls.md). +### 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. + ## [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. diff --git a/awa-model/migrations/v023_install_queue_storage_substrate.sql b/awa-model/migrations/v023_install_queue_storage_substrate.sql index 3f0957937..a003668ec 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 @@ -696,6 +698,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 +724,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 +775,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; 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/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 545898877..e6aed6454 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -21155,7 +21155,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"; @@ -21299,6 +21305,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 @@ -21475,6 +21592,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/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. From 77229ec486de1e46145ab6837fe9ceff24896146 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Mon, 9 Nov 2026 13:03:08 +1300 Subject: [PATCH 2/6] docs(0.6): say compat batch inserts hold a lane's enqueue lock until the transaction ends The 0.6.15 release notes, the 0.6.14 to 0.6.15 upgrade guide, the lane contention troubleshooting entry and the insert module comment said `awa.insert_jobs_compat_batch` holds a lane's enqueue lock only while it writes that lane, or only for the write and the commit. The lock is transaction-scoped: the batch takes it when it reaches that lane, no longer at its first ready row, and it stays held until the calling transaction commits or rolls back, including any work the caller does after the batch. --- CHANGELOG.md | 4 ++-- awa-model/src/insert.rs | 5 +++-- awa/tests/insert_many_batch_test.rs | 6 +++--- docs/troubleshooting.md | 2 +- docs/upgrade-0.6.14-to-0.6.15.md | 2 +- 5 files changed, 10 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ee8834e95..76211cfe6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,7 +19,7 @@ Notable changes between releases. Detailed migration notes for storage transitio ## [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 @@ -48,7 +48,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/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/tests/insert_many_batch_test.rs b/awa/tests/insert_many_batch_test.rs index 63c36b676..546043817 100644 --- a/awa/tests/insert_many_batch_test.rs +++ b/awa/tests/insert_many_batch_test.rs @@ -2,9 +2,9 @@ //! `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. 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: From 8378ca4d8b7baf80f70f06a00946c87c16ea259b Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Mon, 9 Nov 2026 13:05:10 +1300 Subject: [PATCH 3/6] test(0.6): cover the batch insert function's failing unique-conflict mode `awa.insert_jobs_compat_batch` documents that with `p_skip_unique_conflicts = false` a unique conflict fails the whole batch with a unique violation. The Rust entry points pass `true` whenever a batch carries a unique key, so nothing exercised that mode with a conflict. The new test calls the function directly on queue storage and on canonical storage, with a key that is already claimed and with a key repeated inside the batch. Each fails with `23505` and leaves no row behind; the same batch with skipping inserts the rest. --- awa/tests/insert_many_batch_test.rs | 149 ++++++++++++++++++++++++++++ 1 file changed, 149 insertions(+) diff --git a/awa/tests/insert_many_batch_test.rs b/awa/tests/insert_many_batch_test.rs index 546043817..b7446cd12 100644 --- a/awa/tests/insert_many_batch_test.rs +++ b/awa/tests/insert_many_batch_test.rs @@ -10,6 +10,7 @@ //! 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 From 6f87d5388f44d0b9a431091337ed4ac1696211e1 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Mon, 9 Nov 2026 13:09:07 +1300 Subject: [PATCH 4/6] fix(0.6): reset claim batch index EXECUTE before handing the function to the table owner `awa.refresh_claim_batch_job_index_helper` transferred `ensure_claim_batch_job_index` to the owner of `lease_claim_batches` and then revoked `EXECUTE` from everyone but the owner and granted it to the roles holding `TRUNCATE`. A migration role that owned the function but was a member of the table owner only `WITH INHERIT FALSE, SET TRUE` (PostgreSQL 16+) could make the transfer, but afterwards held neither ownership nor grant options, so the revoke and grant only raised warnings and `PUBLIC` kept `EXECUTE`. The helper now resets `EXECUTE` while the running role still owns the function, then transfers ownership; `ALTER FUNCTION ... OWNER TO` keeps the ACL. Running the patch with the schema owner as the effective role, as the database-roles guide says, already produced the right ACL; the new test covers both that and the transfer path with a SET-only membership. --- CHANGELOG.md | 1 + .../v023_install_queue_storage_substrate.sql | 44 ++++--- awa/tests/lane_ownership_roles_test.rs | 116 ++++++++++++++++++ 3 files changed, 141 insertions(+), 20 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 76211cfe6..d4ba7e246 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,7 @@ 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` to the roles holding `TRUNCATE` on `lease_claim_batches` before the ownership transfer. Migrations run with the owner as the effective role, as documented, were not affected. ## [0.6.15] — 2026-11-09 diff --git a/awa-model/migrations/v023_install_queue_storage_substrate.sql b/awa-model/migrations/v023_install_queue_storage_substrate.sql index a003668ec..9bf4e8bad 100644 --- a/awa-model/migrations/v023_install_queue_storage_substrate.sql +++ b/awa-model/migrations/v023_install_queue_storage_substrate.sql @@ -791,6 +791,30 @@ BEGIN v_function := format('%I.ensure_claim_batch_job_index(integer)', p_schema)::regprocedure; + -- Reset EXECUTE before handing the function to the table owner: the + -- running role holds the current owner's privileges (it just replaced the + -- function), but a role that may only SET ROLE to the table owner cannot + -- revoke or grant once that role owns it. ALTER ... OWNER keeps the ACL. + FOR v_grantee IN + SELECT DISTINCT CASE WHEN acl.grantee = 0 THEN 'PUBLIC' + ELSE quote_ident(pg_get_userbyid(acl.grantee)) END + FROM pg_proc AS routine + CROSS JOIN LATERAL aclexplode(COALESCE(routine.proacl, acldefault('f', routine.proowner))) AS acl + WHERE routine.oid = v_function + AND acl.privilege_type = 'EXECUTE' + AND acl.grantee <> routine.proowner + LOOP + EXECUTE format('REVOKE EXECUTE ON FUNCTION %s FROM %s', v_function, v_grantee); + END LOOP; + FOR v_grantee IN + SELECT DISTINCT CASE WHEN acl.grantee = 0 THEN 'PUBLIC' + ELSE quote_ident(pg_get_userbyid(acl.grantee)) END + FROM aclexplode(COALESCE(v_table_acl, acldefault('r', v_table_owner))) AS acl + WHERE acl.privilege_type = 'TRUNCATE' + LOOP + EXECUTE format('GRANT EXECUTE ON FUNCTION %s TO %s', v_function, v_grantee); + END LOOP; + SELECT routine.proowner INTO v_function_owner FROM pg_proc AS routine WHERE routine.oid = v_function; @@ -814,26 +838,6 @@ BEGIN ); END IF; END IF; - - FOR v_grantee IN - SELECT DISTINCT CASE WHEN acl.grantee = 0 THEN 'PUBLIC' - ELSE quote_ident(pg_get_userbyid(acl.grantee)) END - FROM pg_proc AS routine - CROSS JOIN LATERAL aclexplode(COALESCE(routine.proacl, acldefault('f', routine.proowner))) AS acl - WHERE routine.oid = v_function - AND acl.privilege_type = 'EXECUTE' - AND acl.grantee <> routine.proowner - LOOP - EXECUTE format('REVOKE EXECUTE ON FUNCTION %s FROM %s', v_function, v_grantee); - END LOOP; - FOR v_grantee IN - SELECT DISTINCT CASE WHEN acl.grantee = 0 THEN 'PUBLIC' - ELSE quote_ident(pg_get_userbyid(acl.grantee)) END - FROM aclexplode(COALESCE(v_table_acl, acldefault('r', v_table_owner))) AS acl - WHERE acl.privilege_type = 'TRUNCATE' - LOOP - EXECUTE format('GRANT EXECUTE ON FUNCTION %s TO %s', v_function, v_grantee); - END LOOP; END; $refresh$; diff --git a/awa/tests/lane_ownership_roles_test.rs b/awa/tests/lane_ownership_roles_test.rs index f7b85eea0..7c7dcacd6 100644 --- a/awa/tests/lane_ownership_roles_test.rs +++ b/awa/tests/lane_ownership_roles_test.rs @@ -893,6 +893,122 @@ async fn set_only_schema_owner_membership_repairs_legacy_lanes_and_restores_role cluster.teardown().await } +/// `(owner, PUBLIC executes, runtime group executes)` of the default schema's +/// `ensure_claim_batch_job_index`. +async fn claim_batch_job_index_function_acl( + cluster: &Cluster, +) -> Result<(String, bool, bool), sqlx::Error> { + sqlx::query_as( + "SELECT pg_get_userbyid(routine.proowner)::text, \ + has_function_privilege('public', routine.oid, 'EXECUTE'), \ + has_function_privilege($1, routine.oid, 'EXECUTE') \ + FROM pg_proc AS routine \ + WHERE routine.oid = to_regprocedure('awa.ensure_claim_batch_job_index(integer)')", + ) + .bind(&cluster.runtime_name) + .fetch_one(&cluster.owner) + .await +} + +/// A migration role whose membership in the schema owner is SET-only +/// (PostgreSQL 16+: it may `SET ROLE` to the owner but does not inherit its +/// privileges) leaves `ensure_claim_batch_job_index` owned by the table owner, +/// executable by the roles that hold `TRUNCATE` on `lease_claim_batches` and +/// not by `PUBLIC`: when it refreshes the function as itself, handing a +/// function it owns to the owner, and when it applies the +/// `claim_batch_job_index` patch as the owner, as the database-roles guide +/// says to. +#[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; + } + assert_eq!( + claim_batch_job_index_function_acl(&cluster).await?, + (cluster.owner_name.clone(), false, true) + ); + + // A function the migration role owns, executable by PUBLIC, as a refresh + // by the migration role acting as itself would create it. + cluster + .owner + .execute(sqlx::raw_sql(&sql(format!( + "GRANT USAGE, CREATE ON SCHEMA awa TO {migrator}; \ + GRANT EXECUTE ON FUNCTION awa.refresh_claim_batch_job_index_helper(text) TO {migrator}", + migrator = cluster.migrator_name + )))) + .await?; + 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?; + 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 (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(&cluster.migrator) + .await?; + assert!(!inherits_owner); + assert!(can_set_owner); + + sqlx::query("SELECT awa.refresh_claim_batch_job_index_helper('awa')") + .execute(&cluster.migrator) + .await?; + assert_eq!( + claim_batch_job_index_function_acl(&cluster).await?, + (cluster.owner_name.clone(), false, true), + "a refresh by the migration role acting as itself" + ); + + // 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?; + migrate_pool.close().await; + 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?, + (cluster.owner_name.clone(), false, true), + "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 From be1aa3cad3fd3e3d6376da9257a78aa9a259f03a Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Mon, 9 Nov 2026 13:12:40 +1300 Subject: [PATCH 5/6] test(0.6): count the rows each insert returns in the lane contention benchmark The benchmark added the batch size to its job count for every batch, so a skipping insert (`insert_many_skip_unique_conflicts`, `insert_many_copy`) that left out a duplicate still counted it, inflating the reported rate, and the integrity check only asked that at least the measured window's count had landed. Each mode now reports the rows its insert returned, the row-at-a-time fallback counts the rows that went in, and throughput uses those. The output shows inserted against requested jobs, the JSON record carries both, and the integrity check requires the rows that landed to equal the jobs the inserts returned over the whole run. --- awa/tests/enqueue_lane_contention_test.rs | 73 +++++++++++++++-------- 1 file changed, 47 insertions(+), 26 deletions(-) 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 ); } From 709f8c440df6ec697cc057d939e8799b31cccf63 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Mon, 9 Nov 2026 13:49:29 +1300 Subject: [PATCH 6/6] fix(0.6): reset claim batch index EXECUTE as the owner after the ownership transfer Resetting `EXECUTE` before handing `ensure_claim_batch_job_index` to the table owner lost the former owner's access: its grant to itself was its owner entry, and `ALTER FUNCTION ... OWNER TO` rewrites that entry to the new owner. A former owner that holds `TRUNCATE` on `lease_claim_batches`, and runs claim prune, could then no longer execute the function, so its prunes skipped the index build. The helper now transfers ownership first and resets `EXECUTE` as a role with the owner's privileges. A role that may only `SET ROLE` to the table owner (PostgreSQL 16+ `INHERIT FALSE, SET TRUE`) switches to it with a transaction-local `role` setting for the reset and then restores its previous `role` value, so a `role` set through `PGOPTIONS` is kept. The final ACL on every path is the roles holding `TRUNCATE`, including a former owner, and not `PUBLIC`. The SET-only test now gives the migration role `TRUNCATE`, also covers an ordinary member refreshing as itself, and checks the caller's role is unchanged afterwards. --- CHANGELOG.md | 2 +- .../v023_install_queue_storage_substrate.sql | 57 +++++---- awa/tests/lane_ownership_roles_test.rs | 112 ++++++++++++------ 3 files changed, 113 insertions(+), 58 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6dc111a33..dec026559 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,7 +16,7 @@ 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` to the roles holding `TRUNCATE` on `lease_claim_batches` before the ownership transfer. Migrations run with the owner as the effective role, as documented, were not affected. +- 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 diff --git a/awa-model/migrations/v023_install_queue_storage_substrate.sql b/awa-model/migrations/v023_install_queue_storage_substrate.sql index 9bf4e8bad..6a688758b 100644 --- a/awa-model/migrations/v023_install_queue_storage_substrate.sql +++ b/awa-model/migrations/v023_install_queue_storage_substrate.sql @@ -673,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 @@ -791,30 +792,6 @@ BEGIN v_function := format('%I.ensure_claim_batch_job_index(integer)', p_schema)::regprocedure; - -- Reset EXECUTE before handing the function to the table owner: the - -- running role holds the current owner's privileges (it just replaced the - -- function), but a role that may only SET ROLE to the table owner cannot - -- revoke or grant once that role owns it. ALTER ... OWNER keeps the ACL. - FOR v_grantee IN - SELECT DISTINCT CASE WHEN acl.grantee = 0 THEN 'PUBLIC' - ELSE quote_ident(pg_get_userbyid(acl.grantee)) END - FROM pg_proc AS routine - CROSS JOIN LATERAL aclexplode(COALESCE(routine.proacl, acldefault('f', routine.proowner))) AS acl - WHERE routine.oid = v_function - AND acl.privilege_type = 'EXECUTE' - AND acl.grantee <> routine.proowner - LOOP - EXECUTE format('REVOKE EXECUTE ON FUNCTION %s FROM %s', v_function, v_grantee); - END LOOP; - FOR v_grantee IN - SELECT DISTINCT CASE WHEN acl.grantee = 0 THEN 'PUBLIC' - ELSE quote_ident(pg_get_userbyid(acl.grantee)) END - FROM aclexplode(COALESCE(v_table_acl, acldefault('r', v_table_owner))) AS acl - WHERE acl.privilege_type = 'TRUNCATE' - LOOP - EXECUTE format('GRANT EXECUTE ON FUNCTION %s TO %s', v_function, v_grantee); - END LOOP; - SELECT routine.proowner INTO v_function_owner FROM pg_proc AS routine WHERE routine.oid = v_function; @@ -828,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 @@ -838,6 +816,37 @@ 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 + FROM pg_proc AS routine + CROSS JOIN LATERAL aclexplode(COALESCE(routine.proacl, acldefault('f', routine.proowner))) AS acl + WHERE routine.oid = v_function + AND acl.privilege_type = 'EXECUTE' + AND acl.grantee <> routine.proowner + LOOP + EXECUTE format('REVOKE EXECUTE ON FUNCTION %s FROM %s', v_function, v_grantee); + END LOOP; + FOR v_grantee IN + SELECT DISTINCT CASE WHEN acl.grantee = 0 THEN 'PUBLIC' + ELSE quote_ident(pg_get_userbyid(acl.grantee)) END + FROM aclexplode(COALESCE(v_table_acl, acldefault('r', v_table_owner))) AS acl + WHERE acl.privilege_type = 'TRUNCATE' + 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/tests/lane_ownership_roles_test.rs b/awa/tests/lane_ownership_roles_test.rs index 7c7dcacd6..99a2ec5a8 100644 --- a/awa/tests/lane_ownership_roles_test.rs +++ b/awa/tests/lane_ownership_roles_test.rs @@ -893,31 +893,56 @@ async fn set_only_schema_owner_membership_repairs_legacy_lanes_and_restores_role cluster.teardown().await } -/// `(owner, PUBLIC executes, runtime group executes)` of the default schema's -/// `ensure_claim_batch_job_index`. +/// 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<(String, bool, bool), sqlx::Error> { +) -> Result { sqlx::query_as( - "SELECT pg_get_userbyid(routine.proowner)::text, \ - has_function_privilege('public', routine.oid, 'EXECUTE'), \ - has_function_privilege($1, routine.oid, 'EXECUTE') \ + "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 } -/// A migration role whose membership in the schema owner is SET-only -/// (PostgreSQL 16+: it may `SET ROLE` to the owner but does not inherit its -/// privileges) leaves `ensure_claim_batch_job_index` owned by the table owner, -/// executable by the roles that hold `TRUNCATE` on `lease_claim_batches` and -/// not by `PUBLIC`: when it refreshes the function as itself, handing a -/// function it owns to the owner, and when it applies the -/// `claim_batch_job_index` patch as the owner, as the database-roles guide -/// says to. +/// 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?; @@ -928,29 +953,39 @@ async fn set_only_schema_owner_membership_resets_claim_batch_job_index_execute() 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?, - (cluster.owner_name.clone(), false, true) + expected ); - // A function the migration role owns, executable by PUBLIC, as a refresh - // by the migration role acting as itself would create it. 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?; - 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?; + 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!( @@ -960,22 +995,28 @@ async fn set_only_schema_owner_membership_resets_claim_batch_job_index_execute() 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(&cluster.migrator) + .fetch_one(&mut *conn) .await?; assert!(!inherits_owner); assert!(can_set_owner); - - sqlx::query("SELECT awa.refresh_claim_batch_job_index_helper('awa')") - .execute(&cluster.migrator) - .await?; + 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?, - (cluster.owner_name.clone(), false, true), - "a refresh by the migration role acting as itself" + 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. @@ -993,7 +1034,12 @@ async fn set_only_schema_owner_membership_resets_claim_batch_job_index_execute() ["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 @@ -1003,7 +1049,7 @@ async fn set_only_schema_owner_membership_resets_claim_batch_job_index_execute() assert_eq!(patch.outcome, migrations::PatchOutcome::Applied); assert_eq!( claim_batch_job_index_function_acl(&cluster).await?, - (cluster.owner_name.clone(), false, true), + expected, "the patch applied by the migration role as the owner" ); cluster.teardown().await