Forward-port 0.6.15 to main

.github/workflows/nightly-chaos.yml

@@ -197,6 +197,17 @@ jobs:
           cargo test --release --package awa --test concurrent_lifecycle_test \
             test_concurrent_queue_sweep -- --exact --ignored --nocapture \
             | tee artifacts/nightly/rust-queue-sweep.txt
+      - name: Run per-lane enqueue contention benchmark
+        if: always()
+        timeout-minutes: 5
+        run: |
+          set -o pipefail
+          # Many producers batch-inserting into one queue lane. Gates the
+          # per-lane rate and p99 through benchmarks/baseline.json, and the
+          # lane-lock hold time relative to the per-row insert function.
+          cargo test --release --package awa --test enqueue_lane_contention_test \
+            test_enqueue_lane_contention_gate -- --exact --ignored --nocapture \
+            | tee artifacts/nightly/rust-enqueue-lane.txt
       - name: Run MVCC horizon overlap benchmark
         if: always()
         timeout-minutes: 10

CHANGELOG.md

@@ -8,6 +8,10 @@ Notable changes between releases. Detailed migration notes for storage transitio
 
 - Everything [0.6.14](https://github.com/hardbyte/awa/releases/tag/v0.6.14) shipped that this branch did not already have; see its release notes for the details. Promotion runs each enqueue lane in its own transaction, hands a busy lane to a dedicated waiter, and wakes claimers after the lanes commit; `awa.maintenance.promote.lane_contention` counts the hand-offs. Rescue skips an attempt whose retained ready bodies are ambiguous instead of stopping rescue of every other attempt. Lifecycle events no longer re-read a completed job, and a failed re-read no longer fails a committed callback wait. Admin job lists and kind counts find their candidates in read phases that release partition locks between them, and probe receipt closures more cheaply; a job list still loads each candidate with a point read, which holds `ACCESS SHARE` on every ready, lease and claim partition for the length of that read. `awa migrate` rejects `--sql` combined with `--extract-to`. The persona soak suite (`awa/tests/persona_soak_test.rs`, `scripts/persona-soak.sh`) is added.
 
+### Forward-ported from 0.6.15
+
+- Everything [0.6.15](https://github.com/hardbyte/awa/releases/tag/v0.6.15) shipped that this branch did not already have; see its release notes for the details. Lifecycle hooks hold their attempt's finalizer slot until they return, so hooks slower than job throughput slow claiming instead of growing worker memory, and maintenance keeps at most 1,024 detached `Rescued` and duplicate `Cancelled` hook tasks per worker. By-job lookups of compact receipt claims use a GIN index on `lease_claim_batches.job_ids` that claim prune builds on each partition as it empties it (migration v054). Compatibility COPY inserts go through `awa.insert_jobs_compat_batch` (migration v055), which takes each queue lane's enqueue lock only when it writes that lane rather than at the batch's first ready row (the lock is still held until the transaction ends), and the new `insert_many_skip_unique_conflicts` inserts a batch through it while skipping claimed unique keys. Rescue reaches stale attempts behind a full batch of ambiguous ones, `Completed` events and follow-ups carry the completed job without the attempt's heartbeat and deadline, slot-scoped admin receipt reads stay off other slots' closure partitions, and unfiltered receipt job lists select candidate IDs before loading bodies. v054 and v055 add functions only and take no table locks; a database coming from 0.6.15 already has both functions from its schema patches. See [v054](docs/upgrade-0.6-to-0.7.md#v054-claim-batch-job-index) and [v055](docs/upgrade-0.6-to-0.7.md#v055-batch-compatibility-insert) in the upgrade guide.
+
 ### Added
 
 - **The queue-storage prune `lock_timeout` is configurable, and prune

README.md

@@ -93,9 +93,9 @@ See [docs/positioning.md](docs/positioning.md) for the category map and messagin
 
 ```bash
 # 1. Add Awa to a Python project
-uv add 'awa-pg[ui]==0.6.6'     # Python SDK + dashboard binary
-# uv add awa-pg==0.6.6         # SDK only (no dashboard, smaller wheel)
-# or: cargo add awa@0.6.6      # Rust
+uv add 'awa-pg[ui]'     # Python SDK + dashboard binary
+# uv add awa-pg         # SDK only (no dashboard, smaller wheel)
+# or: cargo add awa      # Rust
 
 # 2. Start Postgres and run migrations
 awa --database-url $DATABASE_URL migrate
@@ -295,7 +295,7 @@ tx.commit().await?;
 awa::admin::cancel_by_unique_key(&pool, "send_email", None, Some(&serde_json::json!({"to": "alice@example.com", "subject": "Welcome"})), None).await?;
 ```
 
-Lifecycle hooks receive `Started` after a claim commits as the worker handler is about to run, plus outcome events (`Completed`, `Retried`, `Exhausted`, `Cancelled`) after the corresponding state transition commits. Hooks run in detached tasks, so they do not block job execution.
+Lifecycle hooks receive `Started` after a claim commits as the worker handler is about to run, plus outcome events (`Completed`, `Retried`, `Exhausted`, `Cancelled`) after the corresponding state transition commits. Hooks run alongside job execution and never delay a job's state transition, but each attempt's hooks hold a finalizer slot until they return, so hooks slower than job throughput slow claiming.
 
 Cancellation is cooperative for running handlers:
 
@@ -310,27 +310,26 @@ Cancellation is cooperative for running handlers:
 ### Python
 
 ```bash
-uv add awa-pg==0.6.6            # SDK: insert, worker, admin, progress
-uv add 'awa-pg[ui]==0.6.6'      # SDK + bundled `awa` binary for the dashboard
+uv add awa-pg            # SDK: insert, worker, admin, progress
+uv add 'awa-pg[ui]'      # SDK + bundled `awa` binary for the dashboard
 # or, just the CLI:
-uv tool install awa-cli==0.6.6  # CLI on its own: migrations, queue admin, web UI
+uv tool install awa-cli  # CLI on its own: migrations, queue admin, web UI
 ```
 
-`uv add awa-pg==0.6.6` stays small for workers and producers. The `[ui]` extra pulls in [`awa-cli`](https://pypi.org/project/awa-cli/), which ships the `awa` binary plus the embedded React dashboard; afterwards `uv run python -m awa serve` (or `awa serve` directly) launches it.
+`uv add awa-pg` stays small for workers and producers. The `[ui]` extra pulls in [`awa-cli`](https://pypi.org/project/awa-cli/), which ships the `awa` binary plus the embedded React dashboard; afterwards `uv run python -m awa serve` (or `awa serve` directly) launches it.
 
 ### Rust
 
-```toml
-[dependencies]
-awa = "0.6"
+```bash
+cargo add awa
 ```
 
 ### CLI
 
 Available as a uv tool (no Rust toolchain needed) or through cargo:
 
 ```bash
-uv tool install awa-cli==0.6.6
+uv tool install awa-cli
 # or: cargo install awa-cli
 
 awa --database-url $DATABASE_URL migrate
@@ -376,7 +375,7 @@ All coordination through Postgres. The Rust runtime owns dispatch, leases, heart
 | `awa-worker`  | Runtime: dispatch, heartbeat, maintenance          |
 | `awa-ui`      | Web UI (axum API + embedded React frontend)        |
 | `awa-cli`     | CLI binary (migrations, admin, serve)              |
-| `awa-python`  | PyO3 extension module (`uv add awa-pg==0.6.6`)    |
+| `awa-python`  | PyO3 extension module (`uv add awa-pg`)           |
 | `awa-testing` | Test helpers (`TestClient`)                        |
 
 ## Documentation

awa-model/migrations/v023_install_queue_storage_substrate.sql

@@ -792,6 +792,177 @@ $job_output$;
 
 REVOKE EXECUTE ON FUNCTION awa.install_queue_storage_job_output(TEXT) FROM PUBLIC;
 
+-- Installs p_schema.ensure_claim_batch_job_index(p_slot), which adds the GIN
+-- index on job_ids to one lease_claim_batches partition when that partition
+-- is empty and its lock is free, and otherwise does nothing. Building the
+-- 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.
+--
+-- 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
+-- pending list does, while a pending list would make every lookup read it
+-- (up to gin_pending_list_limit) and push its flush into a claim insert.
+--
+-- The function is SECURITY DEFINER, owned by the owner of lease_claim_batches
+-- (CREATE INDEX needs table ownership), with a pinned search_path. EXECUTE
+-- goes to the roles that may TRUNCATE lease_claim_batches, which are the roles
+-- that run claim prune, rather than to PUBLIC. Replacing it takes no table
+-- locks and reads no rows. The installer calls this too.
+CREATE OR REPLACE FUNCTION awa.refresh_claim_batch_job_index_helper(p_schema TEXT)
+RETURNS VOID
+LANGUAGE plpgsql
+SECURITY INVOKER
+SET search_path = pg_catalog, pg_temp
+AS $refresh$
+DECLARE
+    v_table_owner OID;
+    v_table_acl ACLITEM[];
+    v_function REGPROCEDURE;
+    v_function_owner OID;
+    v_grantee TEXT;
+    v_set_mode TEXT := CASE WHEN current_setting('server_version_num')::int >= 160000
+                            THEN 'SET' ELSE 'MEMBER' END;
+BEGIN
+    IF p_schema IS NULL OR p_schema !~ '^[a-z_][a-z0-9_]*$' THEN
+        RAISE EXCEPTION 'refresh_claim_batch_job_index_helper: schema name must match ^[a-z_][a-z0-9_]*$; got %',
+            p_schema
+            USING ERRCODE = '22023';
+    END IF;
+
+    SELECT batches.relowner, batches.relacl INTO v_table_owner, v_table_acl
+    FROM pg_class AS batches
+    WHERE batches.oid = to_regclass(format('%I.lease_claim_batches', p_schema));
+    IF v_table_owner IS NULL THEN
+        RAISE EXCEPTION 'refresh_claim_batch_job_index_helper: schema % has no lease_claim_batches table',
+            p_schema
+            USING ERRCODE = '22023';
+    END IF;
+
+    EXECUTE format(
+        $ddl$
+        CREATE OR REPLACE FUNCTION %1$I.ensure_claim_batch_job_index(p_slot INTEGER)
+        RETURNS BOOLEAN
+        LANGUAGE plpgsql
+        SECURITY DEFINER
+        SET search_path = pg_catalog, pg_temp
+        SET quote_all_identifiers = off
+        AS $func$
+        DECLARE
+            v_child TEXT;
+            v_index TEXT;
+            v_has_rows BOOLEAN;
+        BEGIN
+            IF p_slot IS NULL OR p_slot < 0 THEN
+                RAISE EXCEPTION 'invalid claim slot: %%', p_slot
+                    USING ERRCODE = '22023';
+            END IF;
+
+            v_child := format('lease_claim_batches_%%s', p_slot);
+            v_index := format('lease_claim_batches_%%s_job_ids_idx', p_slot);
+
+            IF NOT EXISTS (
+                SELECT 1
+                FROM pg_inherits AS inheritance
+                WHERE inheritance.inhparent = to_regclass(format('%%I.lease_claim_batches', %1$L))
+                  AND inheritance.inhrelid = to_regclass(format('%%I.%%I', %1$L, v_child))
+            ) THEN
+                RAISE EXCEPTION 'schema %% has no lease_claim_batches partition for claim slot %%',
+                    %1$L, p_slot
+                    USING ERRCODE = '22023';
+            END IF;
+
+            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
+                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)%%'
+            ) THEN
+                RETURN TRUE;
+            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
+            -- subtransaction, which releases a lock taken here on a populated
+            -- partition; a lock the caller already held is kept.
+            BEGIN
+                EXECUTE format('LOCK TABLE %%I.%%I IN SHARE MODE NOWAIT', %1$L, v_child);
+                EXECUTE format('SELECT EXISTS (SELECT 1 FROM %%I.%%I)', %1$L, v_child)
+                    INTO v_has_rows;
+                IF v_has_rows THEN
+                    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)',
+                    v_index, %1$L, v_child
+                );
+            EXCEPTION
+                WHEN lock_not_available OR object_not_in_prerequisite_state THEN
+                    RETURN FALSE;
+            END;
+            RETURN TRUE;
+        END;
+        $func$
+        $ddl$,
+        p_schema
+    );
+
+    v_function := format('%I.ensure_claim_batch_job_index(integer)', p_schema)::regprocedure;
+
+    SELECT routine.proowner INTO v_function_owner
+    FROM pg_proc AS routine
+    WHERE routine.oid = v_function;
+
+    IF v_function_owner <> v_table_owner THEN
+        IF pg_has_role(current_user, v_function_owner, 'USAGE')
+           AND pg_has_role(current_user, v_table_owner, v_set_mode)
+           AND has_schema_privilege(v_table_owner, to_regnamespace(quote_ident(p_schema)), 'CREATE') THEN
+            EXECUTE format(
+                'ALTER FUNCTION %s OWNER TO %I',
+                v_function,
+                pg_get_userbyid(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
+                USING HINT = format(
+                    'Run awa migrate as a role that can SET ROLE to %s, or as %s itself.',
+                    quote_ident(pg_get_userbyid(v_table_owner)),
+                    quote_ident(pg_get_userbyid(v_table_owner))
+                );
+        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$;
+
+REVOKE EXECUTE ON FUNCTION awa.refresh_claim_batch_job_index_helper(TEXT) FROM PUBLIC;
+
 CREATE OR REPLACE FUNCTION awa.install_queue_storage_substrate(
     p_schema               TEXT,
     p_queue_slot_count     INT     DEFAULT 16,
@@ -2535,6 +2706,19 @@ BEGIN
         );
     END LOOP;
 
+    PERFORM awa.refresh_claim_batch_job_index_helper(p_schema);
+    -- The by-job lookups of compact claims use a GIN index that is only built
+    -- on an empty partition whose lock is free (see
+    -- awa.refresh_claim_batch_job_index_helper). A partition that is populated
+    -- or busy now gains it after claim prune next empties it.
+    FOR v_slot IN 0..(p_claim_slot_count - 1) LOOP
+        BEGIN
+            EXECUTE format('SELECT %I.ensure_claim_batch_job_index($1)', p_schema) USING v_slot;
+        EXCEPTION WHEN insufficient_privilege THEN
+            RAISE NOTICE 'claim slot % of schema % keeps scanning for job ids: %', v_slot, p_schema, SQLERRM;
+        END;
+    END LOOP;
+
     EXECUTE format(
         'CREATE INDEX IF NOT EXISTS idx_%s_lease_claim_closures_job_run ON %I.lease_claim_closures (job_id, run_lease)',
         p_schema, p_schema

awa-model/migrations/v054_claim_batch_job_index.sql

@@ -0,0 +1,56 @@
+-- v054: Index compact claim batches by job id.
+-- Tracking: https://github.com/hardbyte/awa/pull/684
+-- Point lookups of compact receipt claims by job id (heartbeat and progress
+-- flush, completion fallbacks, receipt close, cancellation, rescue and
+-- load_job) filter lease_claim_batches with job_ids && / @>, which a GIN index
+-- on job_ids serves. The installer now installs
+-- <schema>.ensure_claim_batch_job_index(slot) through
+-- awa.refresh_claim_batch_job_index_helper and builds the index on each empty
+-- claim partition; claim prune calls the function right after it truncates a
+-- slot, so every partition gains the index within one claim-ring cycle.
+--
+-- This file only installs the function in each queue-storage schema; it
+-- builds no index. `awa migrate` and `awa migrate --sql --from 53 --to 54` run
+-- V23_UP first, which defines the helper.
+--
+-- A 0.6.15 database already carries the function from the 0.6
+-- claim_batch_job_index schema patch; replacing it is idempotent.
+--
+-- N-1 binaries never call the function, and their by-job lookups return the
+-- same rows with or without the index. Binaries that carry the indexed
+-- lookups stay correct on an unindexed partition; they scan the claim batches.
+--
+-- Live load: function replacement only. It takes no lock on any awa or
+-- queue-storage table. Run it as the owner of the queue-storage tables, or as
+-- a migrator that can SET ROLE to it, so the function is owned by the table
+-- owner and can create the index.
+
+DO $$
+DECLARE
+    v_schema TEXT;
+BEGIN
+    FOR v_schema IN
+        SELECT discovered.schema_name FROM awa.queue_storage_schemas() AS discovered
+    LOOP
+        IF NOT EXISTS (
+            SELECT 1 FROM pg_class AS batches
+            WHERE batches.oid = to_regclass(format('%I.lease_claim_batches', v_schema))
+              AND batches.relkind = 'p'
+        ) THEN
+            CONTINUE;
+        END IF;
+
+        -- Serialize with a concurrent prepare_schema() of the same schema. This
+        -- is an advisory lock, not a table lock: claims and completions do not
+        -- take it.
+        PERFORM pg_advisory_xact_lock(
+            hashtextextended(format('awa.queue_storage.install:%I', v_schema), 0)
+        );
+        PERFORM awa.refresh_claim_batch_job_index_helper(v_schema);
+    END LOOP;
+END;
+$$;
+
+INSERT INTO awa.schema_version (version, description)
+VALUES (54, 'Index compact claim batches by job id')
+ON CONFLICT (version) DO NOTHING;