fix(0.6): claim batch index check, SET-only migrator grants, and review follow-ups

CHANGELOG.md

@@ -7,18 +7,21 @@ 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.
+- 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

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$;
 

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

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))"
     };
 }