diff --git a/CHANGELOG.md b/CHANGELOG.md index 8b4fa7fed..41880eab4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ Notable changes between releases. Detailed migration notes for storage transitio - 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. +- Run `awa migrate` once as the schema owner, before or after rolling binaries. It replaces the body of `awa.insert_jobs_compat_batch`, which a 0.6.15 database now reports as the pending `insert_jobs_compat_batch` patch, and takes no table locks. The fix is in the function, so 0.6.15 producers get it as soon as the patch commits. External migration runners: replace `R__insert_jobs_compat_batch.sql` along with the rest of the `R__` set from `awa migrate --extract-to --from 40`. See the [upgrade guide](docs/upgrade-0.6.15-to-0.6.16.md). ### Added @@ -15,6 +16,7 @@ Notable changes between releases. Detailed migration notes for storage transitio ### Fixed +- `insert_many_copy` and `insert_many_skip_unique_conflicts` no longer deadlock with per-row producers (generic `insert_many`, single-job inserts, 0.6.14 and earlier binaries) that share their unique keys or lanes. Since 0.6.15, when a per-row transaction held a lane and then inserted a key a waiting batch had claimed, PostgreSQL aborted one of the two with `deadlock detected` (`40P01`). The batch function now waits at most 500 ms (or half of its session's `deadlock_timeout`, if lower) while it holds claims or lanes: on a longer wait it releases them and tries again, so the per-row transaction proceeds and the batch then treats the keys it committed as already claimed. The same applies to a transaction that enqueues lanes in separate statements in the opposite order to a batch. This covers per-row sessions whose `deadlock_timeout` is above 500 ms, as PostgreSQL's default of 1 s is, and batches whose unique claims insert within that budget (10,000 to 20,000 keys on a fast server, fewer on a slow or busy one); larger batches behave as in 0.6.15. The protection lasts for 2 s of retries; then the batch makes one last attempt with ordinary lock waits, as in 0.6.15. A per-row transaction that holds a lane for longer than that before it inserts one of the batch's keys can therefore still deadlock with it, as in 0.6.15. A cycle through a lock the batch's own transaction took in an earlier statement also ends in `deadlock detected`, after about 3 to 4 s instead of 1 to 2 s, and usually aborts the batch's transaction. The caller's `lock_timeout` applies to each lock wait, as before. Results and the lane lock window are unchanged. - 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. diff --git a/README.md b/README.md index ff0afa109..2995d4c7a 100644 --- a/README.md +++ b/README.md @@ -390,6 +390,7 @@ All coordination through Postgres. The Rust runtime owns dispatch, leases, heart | [0.5 → 0.6 upgrade](docs/upgrade-0.5-to-0.6.md) | Step-by-step operator checklist for the staged storage transition | | [0.6.9 → 0.6.10 upgrade](docs/upgrade-0.6.9-to-0.6.10.md) | Rolling patch upgrade, rollback, and optional runtime privilege tightening | | [0.6.14 → 0.6.15 upgrade](docs/upgrade-0.6.14-to-0.6.15.md) | Rolling patch upgrade: bounded worker finalization, the claim batch index and the batch compatibility insert | +| [0.6.15 → 0.6.16 upgrade](docs/upgrade-0.6.15-to-0.6.16.md) | Rolling patch upgrade: operator prune halt, and batch compatibility inserts that give way to per-row producers instead of deadlocking | | [Configuration reference](docs/configuration.md) | `QueueConfig`, `ClientBuilder`, Python `start()`, env vars | | [HTTP workers and callbacks](docs/http-callbacks.md) | Serverless dispatch, callback receiver endpoints, signature verification | | [Lifecycle hooks](docs/lifecycle-hooks.md) | React to job start/finish/retry/callback events for metrics, alerting, cleanup | diff --git a/awa-model/migrations/patches/insert_jobs_compat_batch.sql b/awa-model/migrations/patches/insert_jobs_compat_batch.sql index 665537a56..06cdf0d1d 100644 --- a/awa-model/migrations/patches/insert_jobs_compat_batch.sql +++ b/awa-model/migrations/patches/insert_jobs_compat_batch.sql @@ -13,6 +13,20 @@ -- segments in two statements. The lock is held across the reservation, those -- two statements and the commit. -- +-- Per-row producers take each job's claim and its lane together, so one can +-- hold a lane the batch needs while it waits for a key the batch has claimed. +-- The batch therefore takes its claims and lanes in a subtransaction, and +-- limits each lock wait to what remains of a budget measured from the +-- attempt's start: 500 ms, or half of the batch session's deadlock_timeout +-- when that is lower. When a wait would run +-- longer, the batch rolls back, which releases every claim and lane it took, +-- and retries, so a per-row producer whose deadlock_timeout exceeds the budget +-- (PostgreSQL's default is 1 s) usually proceeds before its deadlock check +-- runs. After four budgets of retries the batch makes one last attempt with +-- ordinary waits, as it did before it had a budget, so contention or a cycle +-- through locks the caller took earlier ends in success, a lock timeout or +-- deadlock detected, never in endless retries. +-- -- The column arrays are parallel and must have equal length. Every element is -- required except run_at, unique_key, unique_states and ordering_hash; -- insert_job_compat's per-row defaults are the caller's to apply. @@ -71,6 +85,19 @@ DECLARE v_ready_generation BIGINT; v_row RECORD; v_shards SMALLINT[]; + v_result awa.jobs[]; + v_claim RECORD; + v_attempt INT := 0; + v_attempt_started TIMESTAMPTZ; + v_wait_budget INTERVAL; + v_caller_lock_timeout TEXT; + v_caller_wait_ms NUMERIC; + v_wait_ms NUMERIC; + v_wait_is_callers BOOLEAN := FALSE; + v_phase TEXT; + v_claims_one_by_one BOOLEAN := FALSE; + v_bounded_until TIMESTAMPTZ; + v_final BOOLEAN := FALSE; BEGIN IF v_count = 0 THEN RETURN; @@ -237,8 +264,6 @@ BEGIN v_old_search_path := current_setting('search_path'); PERFORM set_config('search_path', format('%I,awa,public', v_schema), true); - -- Everything up to the lane reservation is lane-independent and runs - -- without a lane lock. EXECUTE format( 'SELECT array_agg(nextval(%L::regclass)::bigint) FROM generate_series(1, $1)', format('%I.job_id_seq', v_schema) @@ -266,188 +291,324 @@ BEGIN WITH ORDINALITY AS input(queue, state, ordering_hash, ord) LEFT JOIN awa.queue_meta AS meta ON meta.queue = input.queue; - -- Claims go in unique-key order, so two producers with overlapping keys - -- cannot wait on each other in opposite orders. - IF p_skip_unique_conflicts THEN - WITH claimed AS ( - INSERT INTO awa.job_unique_claims (unique_key, job_id) - SELECT input.unique_key, v_job_ids[input.ord] - FROM unnest(p_unique_key, p_unique_states, p_state) - WITH ORDINALITY AS input(unique_key, unique_states, state, ord) - WHERE input.unique_key IS NOT NULL - AND input.unique_states IS NOT NULL - AND awa.job_state_in_bitmask(input.unique_states::bit(8), input.state::awa.job_state) - ORDER BY input.unique_key, input.ord - ON CONFLICT (unique_key) DO NOTHING - RETURNING job_id - ) - SELECT COALESCE(array_agg(wanted.job_id), ARRAY[]::BIGINT[]) - INTO v_skipped - FROM ( - SELECT v_job_ids[input.ord] AS job_id - FROM unnest(p_unique_key, p_unique_states, p_state) - WITH ORDINALITY AS input(unique_key, unique_states, state, ord) - WHERE input.unique_key IS NOT NULL - AND input.unique_states IS NOT NULL - AND awa.job_state_in_bitmask(input.unique_states::bit(8), input.state::awa.job_state) - ) AS wanted - WHERE NOT EXISTS (SELECT 1 FROM claimed WHERE claimed.job_id = wanted.job_id); - ELSE - INSERT INTO awa.job_unique_claims (unique_key, job_id) - SELECT input.unique_key, v_job_ids[input.ord] - FROM unnest(p_unique_key, p_unique_states, p_state) - WITH ORDINALITY AS input(unique_key, unique_states, state, ord) - WHERE input.unique_key IS NOT NULL - AND input.unique_states IS NOT NULL - AND awa.job_state_in_bitmask(input.unique_states::bit(8), input.state::awa.job_state) - ORDER BY input.unique_key, input.ord; - END IF; + -- Every claim and lane the batch takes is taken inside one attempt, a + -- subtransaction. Until the bounded phase ends, each lock wait in an + -- attempt ends within the wait budget of the attempt's start; a wait that + -- would run longer rolls the attempt back, which releases everything it + -- took, and the batch tries again. A per-row producer that holds a lane + -- and waits for one of the batch's keys is then released before its own + -- deadlock check runs, provided its deadlock_timeout exceeds the budget. + -- The budget is capped because that check runs on the other session's + -- deadlock_timeout, which this session cannot see. + -- + -- Rolling back cannot release what the caller's transaction took before + -- this call, so retrying alone could spin forever on a cycle through such + -- a lock. After the bounded phase the batch makes one final attempt with + -- the caller's lock_timeout and ordinary waits, as the function did before + -- it had a budget: it succeeds, or PostgreSQL's deadlock detector or the + -- caller's lock_timeout ends it. + -- + -- The caller's lock_timeout applies to each wait, as it does to any + -- statement: a bounded wait is the shorter of the remaining budget and the + -- caller's lock_timeout, and only a wait that the caller's lock_timeout + -- ended raises lock_not_available to the caller. The budget is per wait + -- too: one claim statement can wait on several in-flight keys, and a lane + -- reservation can wait for the lane's provisioning lock and then its lane + -- lock, each up to the remaining budget. + v_caller_lock_timeout := current_setting('lock_timeout'); + v_caller_wait_ms := ceil(1000 * extract(epoch FROM v_caller_lock_timeout::interval)); + v_wait_budget := LEAST( + current_setting('deadlock_timeout')::interval / 2, + interval '500 milliseconds' + ); + v_bounded_until := clock_timestamp() + 4 * v_wait_budget; - -- Re-enqueues that are not yet due carry no lane. - INSERT INTO deferred_jobs ( - job_id, kind, queue, args, state, priority, attempt, run_lease, - max_attempts, run_at, attempted_at, finalized_at, created_at, - unique_key, unique_states, payload - ) - SELECT - v_job_ids[input.ord], - input.kind, - input.queue, - input.args, - input.state::awa.job_state, - input.priority, - 0, - 0, - input.max_attempts, - COALESCE(input.run_at, v_now), - NULL, - NULL, - v_now, - input.unique_key, - input.unique_states, - jsonb_build_object( - 'metadata', input.metadata, - 'tags', input.tags, - 'errors', '[]'::jsonb, - 'progress', NULL - ) - FROM unnest( - p_kind, p_queue, p_args, p_state, p_priority, p_max_attempts, p_run_at, - p_metadata, p_tags, p_unique_key, p_unique_states - ) WITH ORDINALITY AS input( - kind, queue, args, state, priority, max_attempts, run_at, - metadata, tags, unique_key, unique_states, ord + -- The result rows do not depend on claims or lanes, so they are built once, + -- before any lock, and the skipped rows are filtered out on return. + SELECT array_agg( + ( + v_job_ids[input.ord], + input.kind, + input.queue, + input.args, + input.state::awa.job_state, + input.priority, + 0::smallint, + input.max_attempts, + COALESCE(input.run_at, v_now), + NULL::timestamptz, + NULL::timestamptz, + NULL::timestamptz, + NULL::timestamptz, + v_now, + ARRAY[]::jsonb[], + input.metadata, + ARRAY( + SELECT tag.value + FROM jsonb_array_elements_text(input.tags) WITH ORDINALITY AS tag(value, tag_ord) + ORDER BY tag.tag_ord + ), + input.unique_key, + input.unique_states::bit(8), + NULL::uuid, + NULL::timestamptz, + NULL::text, + NULL::text, + NULL::text, + NULL::text, + 0::bigint, + NULL::jsonb + )::awa.jobs + ORDER BY input.ord ) - WHERE input.state <> 'available' - AND v_job_ids[input.ord] <> ALL (v_skipped); - - -- The result rows do not depend on the lanes, so build them before the - -- lock window opens. - RETURN QUERY - SELECT - v_job_ids[input.ord], - input.kind, - input.queue, - input.args, - input.state::awa.job_state, - input.priority, - 0::smallint, - input.max_attempts, - COALESCE(input.run_at, v_now), - NULL::timestamptz, - NULL::timestamptz, - NULL::timestamptz, - NULL::timestamptz, - v_now, - ARRAY[]::jsonb[], - input.metadata, - ARRAY( - SELECT tag.value - FROM jsonb_array_elements_text(input.tags) WITH ORDINALITY AS tag(value, tag_ord) - ORDER BY tag.tag_ord - ), - input.unique_key, - input.unique_states::bit(8), - NULL::uuid, - NULL::timestamptz, - NULL::text, - NULL::text, - NULL::text, - NULL::text, - 0::bigint, - NULL::jsonb + INTO v_result FROM unnest( p_kind, p_queue, p_args, p_state, p_priority, p_max_attempts, p_run_at, p_metadata, p_tags, p_unique_key, p_unique_states ) WITH ORDINALITY AS input( kind, queue, args, state, priority, max_attempts, run_at, metadata, tags, unique_key, unique_states, ord - ) - WHERE v_job_ids[input.ord] <> ALL (v_skipped) - ORDER BY input.ord; + ); - -- Lane lock window: in (queue, priority, shard) order, register each lane - -- the batch is the first to use and reserve its sequence range once, then - -- write the ready rows. The locks are held to the end of the transaction, - -- which keeps the rows of a lane visible in lane_seq order. - -- - -- A new lane is registered only when the loop reaches it, as every other - -- producer does: the head tables' row trigger provisions the lane under - -- its lock, and the uncommitted head row blocks other producers of that - -- lane, so registering a later lane before reserving an earlier one would - -- hold it out of order. Only missing rows are proposed, because the - -- trigger fires before the conflict check. - FOR v_lane IN - SELECT lanes.queue, lanes.priority, lanes.shard, lanes.row_count - FROM ( - SELECT input.queue, input.priority, v_shards[input.ord] AS shard, - count(*) AS row_count - FROM unnest(p_queue, p_priority, p_state) - WITH ORDINALITY AS input(queue, priority, state, ord) - WHERE input.state = 'available' - AND v_job_ids[input.ord] <> ALL (v_skipped) - GROUP BY input.queue, input.priority, v_shards[input.ord] - ) AS lanes - ORDER BY lanes.queue COLLATE "C", lanes.priority, lanes.shard LOOP - INSERT INTO queue_lanes (queue, priority) - SELECT v_lane.queue, v_lane.priority - WHERE NOT EXISTS ( - SELECT 1 FROM queue_lanes AS existing - WHERE existing.queue = v_lane.queue AND existing.priority = v_lane.priority - ) - ON CONFLICT ON CONSTRAINT queue_lanes_pkey DO NOTHING; - - INSERT INTO queue_enqueue_heads (queue, priority, enqueue_shard) - SELECT v_lane.queue, v_lane.priority, v_lane.shard - WHERE NOT EXISTS ( - SELECT 1 FROM queue_enqueue_heads AS existing - WHERE existing.queue = v_lane.queue - AND existing.priority = v_lane.priority - AND existing.enqueue_shard = v_lane.shard - ) - ON CONFLICT (queue, priority, enqueue_shard) DO NOTHING; - - INSERT INTO queue_claim_heads (queue, priority, enqueue_shard) - SELECT v_lane.queue, v_lane.priority, v_lane.shard - WHERE NOT EXISTS ( - SELECT 1 FROM queue_claim_heads AS existing - WHERE existing.queue = v_lane.queue - AND existing.priority = v_lane.priority - AND existing.enqueue_shard = v_lane.shard - ) - ON CONFLICT (queue, priority, enqueue_shard) DO NOTHING; + v_attempt := v_attempt + 1; + BEGIN + v_skipped := ARRAY[]::BIGINT[]; + v_lane_queues := ARRAY[]::TEXT[]; + v_lane_priorities := ARRAY[]::SMALLINT[]; + v_lane_shards := ARRAY[]::SMALLINT[]; + v_lane_starts := ARRAY[]::BIGINT[]; + v_attempt_started := clock_timestamp(); + + -- Claims go in unique-key order, so two producers with overlapping + -- keys cannot wait on each other in opposite orders. The first + -- attempt inserts them in one statement, each lock wait capped at + -- 10 ms (or the budget, if shorter), and gives way to any claim of + -- the same key still in flight; later bounded attempts go key by + -- key, each waiting at most for the rest of the budget; the final + -- attempt inserts them in one statement with ordinary waits. + v_phase := 'claims'; + IF v_final OR NOT v_claims_one_by_one THEN + v_wait_ms := LEAST(10, ceil(1000 * extract(epoch FROM + v_attempt_started + v_wait_budget - clock_timestamp()))); + v_wait_is_callers := v_final OR v_caller_wait_ms BETWEEN 1 AND v_wait_ms; + PERFORM set_config('lock_timeout', CASE WHEN v_wait_is_callers THEN v_caller_lock_timeout + ELSE format('%sms', GREATEST(1, v_wait_ms)::bigint) END, true); + IF p_skip_unique_conflicts THEN + WITH claimed AS ( + INSERT INTO awa.job_unique_claims (unique_key, job_id) + SELECT input.unique_key, v_job_ids[input.ord] + FROM unnest(p_unique_key, p_unique_states, p_state) + WITH ORDINALITY AS input(unique_key, unique_states, state, ord) + WHERE input.unique_key IS NOT NULL + AND input.unique_states IS NOT NULL + AND awa.job_state_in_bitmask(input.unique_states::bit(8), input.state::awa.job_state) + ORDER BY input.unique_key, input.ord + ON CONFLICT (unique_key) DO NOTHING + RETURNING job_id + ) + SELECT COALESCE(array_agg(wanted.job_id), ARRAY[]::BIGINT[]) + INTO v_skipped + FROM ( + SELECT v_job_ids[input.ord] AS job_id + FROM unnest(p_unique_key, p_unique_states, p_state) + WITH ORDINALITY AS input(unique_key, unique_states, state, ord) + WHERE input.unique_key IS NOT NULL + AND input.unique_states IS NOT NULL + AND awa.job_state_in_bitmask(input.unique_states::bit(8), input.state::awa.job_state) + ) AS wanted + WHERE NOT EXISTS (SELECT 1 FROM claimed WHERE claimed.job_id = wanted.job_id); + ELSE + INSERT INTO awa.job_unique_claims (unique_key, job_id) + SELECT input.unique_key, v_job_ids[input.ord] + FROM unnest(p_unique_key, p_unique_states, p_state) + WITH ORDINALITY AS input(unique_key, unique_states, state, ord) + WHERE input.unique_key IS NOT NULL + AND input.unique_states IS NOT NULL + AND awa.job_state_in_bitmask(input.unique_states::bit(8), input.state::awa.job_state) + ORDER BY input.unique_key, input.ord; + END IF; + ELSE + FOR v_claim IN + SELECT input.unique_key, v_job_ids[input.ord] AS job_id + FROM unnest(p_unique_key, p_unique_states, p_state) + WITH ORDINALITY AS input(unique_key, unique_states, state, ord) + WHERE input.unique_key IS NOT NULL + AND input.unique_states IS NOT NULL + AND awa.job_state_in_bitmask(input.unique_states::bit(8), input.state::awa.job_state) + ORDER BY input.unique_key, input.ord + LOOP + v_wait_ms := ceil(1000 * extract(epoch FROM + v_attempt_started + v_wait_budget - clock_timestamp())); + v_wait_is_callers := v_caller_wait_ms BETWEEN 1 AND v_wait_ms; + PERFORM set_config('lock_timeout', CASE WHEN v_wait_is_callers THEN v_caller_lock_timeout + ELSE format('%sms', GREATEST(1, v_wait_ms)::bigint) END, true); + IF p_skip_unique_conflicts THEN + INSERT INTO awa.job_unique_claims (unique_key, job_id) + VALUES (v_claim.unique_key, v_claim.job_id) + ON CONFLICT (unique_key) DO NOTHING; + IF NOT FOUND THEN + v_skipped := v_skipped || v_claim.job_id; + END IF; + ELSE + INSERT INTO awa.job_unique_claims (unique_key, job_id) + VALUES (v_claim.unique_key, v_claim.job_id); + END IF; + END LOOP; + END IF; + + -- The writes outside the lane loop wait at most for relation + -- extension, or for a prune that gives way under its own + -- lock_timeout, so they keep the caller's lock_timeout. + v_wait_is_callers := TRUE; + PERFORM set_config('lock_timeout', v_caller_lock_timeout, true); - EXECUTE format('SELECT %I.reserve_enqueue_seq($1, $2, $3, $4)', v_schema) - INTO v_start - USING v_lane.queue, v_lane.priority, v_lane.shard, v_lane.row_count::BIGINT; + -- Re-enqueues that are not yet due carry no lane. + INSERT INTO deferred_jobs ( + job_id, kind, queue, args, state, priority, attempt, run_lease, + max_attempts, run_at, attempted_at, finalized_at, created_at, + unique_key, unique_states, payload + ) + SELECT + v_job_ids[input.ord], + input.kind, + input.queue, + input.args, + input.state::awa.job_state, + input.priority, + 0, + 0, + input.max_attempts, + COALESCE(input.run_at, v_now), + NULL, + NULL, + v_now, + input.unique_key, + input.unique_states, + jsonb_build_object( + 'metadata', input.metadata, + 'tags', input.tags, + 'errors', '[]'::jsonb, + 'progress', NULL + ) + FROM unnest( + p_kind, p_queue, p_args, p_state, p_priority, p_max_attempts, p_run_at, + p_metadata, p_tags, p_unique_key, p_unique_states + ) WITH ORDINALITY AS input( + kind, queue, args, state, priority, max_attempts, run_at, + metadata, tags, unique_key, unique_states, ord + ) + WHERE input.state <> 'available' + AND v_job_ids[input.ord] <> ALL (v_skipped); + + -- Lane lock window: in (queue, priority, shard) order, register each + -- lane the batch is the first to use and reserve its sequence + -- range once, then write the ready rows. The locks are held to the + -- end of the transaction, which keeps the rows of a lane visible + -- in lane_seq order. + -- + -- A new lane is registered only when the loop reaches it, as every + -- other producer does: the head tables' row trigger provisions the + -- lane under its lock, and the uncommitted head row blocks other + -- producers of that lane, so registering a later lane before + -- reserving an earlier one would hold it out of order. Only missing + -- rows are proposed, because the trigger fires before the conflict + -- check. + v_phase := 'lanes'; + FOR v_lane IN + SELECT lanes.queue, lanes.priority, lanes.shard, lanes.row_count + FROM ( + SELECT input.queue, input.priority, v_shards[input.ord] AS shard, + count(*) AS row_count + FROM unnest(p_queue, p_priority, p_state) + WITH ORDINALITY AS input(queue, priority, state, ord) + WHERE input.state = 'available' + AND v_job_ids[input.ord] <> ALL (v_skipped) + GROUP BY input.queue, input.priority, v_shards[input.ord] + ) AS lanes + ORDER BY lanes.queue COLLATE "C", lanes.priority, lanes.shard + LOOP + v_wait_ms := ceil(1000 * extract(epoch FROM + v_attempt_started + v_wait_budget - clock_timestamp())); + v_wait_is_callers := v_final OR v_caller_wait_ms BETWEEN 1 AND v_wait_ms; + PERFORM set_config('lock_timeout', CASE WHEN v_wait_is_callers THEN v_caller_lock_timeout + ELSE format('%sms', GREATEST(1, v_wait_ms)::bigint) END, true); + + INSERT INTO queue_lanes (queue, priority) + SELECT v_lane.queue, v_lane.priority + WHERE NOT EXISTS ( + SELECT 1 FROM queue_lanes AS existing + WHERE existing.queue = v_lane.queue AND existing.priority = v_lane.priority + ) + ON CONFLICT ON CONSTRAINT queue_lanes_pkey DO NOTHING; + + v_wait_ms := ceil(1000 * extract(epoch FROM + v_attempt_started + v_wait_budget - clock_timestamp())); + v_wait_is_callers := v_final OR v_caller_wait_ms BETWEEN 1 AND v_wait_ms; + PERFORM set_config('lock_timeout', CASE WHEN v_wait_is_callers THEN v_caller_lock_timeout + ELSE format('%sms', GREATEST(1, v_wait_ms)::bigint) END, true); + + INSERT INTO queue_enqueue_heads (queue, priority, enqueue_shard) + SELECT v_lane.queue, v_lane.priority, v_lane.shard + WHERE NOT EXISTS ( + SELECT 1 FROM queue_enqueue_heads AS existing + WHERE existing.queue = v_lane.queue + AND existing.priority = v_lane.priority + AND existing.enqueue_shard = v_lane.shard + ) + ON CONFLICT (queue, priority, enqueue_shard) DO NOTHING; + + v_wait_ms := ceil(1000 * extract(epoch FROM + v_attempt_started + v_wait_budget - clock_timestamp())); + v_wait_is_callers := v_final OR v_caller_wait_ms BETWEEN 1 AND v_wait_ms; + PERFORM set_config('lock_timeout', CASE WHEN v_wait_is_callers THEN v_caller_lock_timeout + ELSE format('%sms', GREATEST(1, v_wait_ms)::bigint) END, true); + + INSERT INTO queue_claim_heads (queue, priority, enqueue_shard) + SELECT v_lane.queue, v_lane.priority, v_lane.shard + WHERE NOT EXISTS ( + SELECT 1 FROM queue_claim_heads AS existing + WHERE existing.queue = v_lane.queue + AND existing.priority = v_lane.priority + AND existing.enqueue_shard = v_lane.shard + ) + ON CONFLICT (queue, priority, enqueue_shard) DO NOTHING; - v_lane_queues := v_lane_queues || v_lane.queue; - v_lane_priorities := v_lane_priorities || v_lane.priority; - v_lane_shards := v_lane_shards || v_lane.shard; - v_lane_starts := v_lane_starts || v_start; + v_wait_ms := ceil(1000 * extract(epoch FROM + v_attempt_started + v_wait_budget - clock_timestamp())); + v_wait_is_callers := v_final OR v_caller_wait_ms BETWEEN 1 AND v_wait_ms; + PERFORM set_config('lock_timeout', CASE WHEN v_wait_is_callers THEN v_caller_lock_timeout + ELSE format('%sms', GREATEST(1, v_wait_ms)::bigint) END, true); + + EXECUTE format('SELECT %I.reserve_enqueue_seq($1, $2, $3, $4)', v_schema) + INTO v_start + USING v_lane.queue, v_lane.priority, v_lane.shard, v_lane.row_count::BIGINT; + + v_lane_queues := v_lane_queues || v_lane.queue; + v_lane_priorities := v_lane_priorities || v_lane.priority; + v_lane_shards := v_lane_shards || v_lane.shard; + v_lane_starts := v_lane_starts || v_start; + END LOOP; + EXCEPTION WHEN lock_not_available THEN + IF v_wait_is_callers THEN + RAISE; + END IF; + IF v_phase = 'claims' THEN + v_claims_one_by_one := TRUE; + END IF; + v_final := clock_timestamp() >= v_bounded_until; + PERFORM pg_sleep( + LEAST(0.1, 0.001 * power(2, LEAST(v_attempt, 7))) * (0.5 + random() / 2) + ); + CONTINUE; + END; + EXIT; END LOOP; + PERFORM set_config('lock_timeout', v_caller_lock_timeout, true); + + -- The lane locks are held, so the ready rows can be written outside the + -- attempt. IF cardinality(v_lane_queues) > 0 THEN IF to_regprocedure(format('%I.ring_cursor(text)', v_schema)) IS NOT NULL THEN EXECUTE format('SELECT slot, generation FROM %I.ring_cursor(''queue'')', v_schema) @@ -557,7 +718,13 @@ BEGIN GROUP BY runs.queue, runs.priority, runs.shard, runs.run_at, runs.run_id ON CONFLICT (ready_slot, ready_generation, queue, priority, enqueue_shard, first_lane_seq) DO NOTHING; + END IF; + RETURN QUERY + SELECT * FROM unnest(v_result) AS result + WHERE result.id <> ALL (v_skipped); + + IF cardinality(v_lane_queues) > 0 THEN PERFORM pg_notify('awa:' || notified.queue, '') FROM (SELECT DISTINCT unnest(v_lane_queues) AS queue) AS notified; END IF; diff --git a/awa-model/src/insert.rs b/awa-model/src/insert.rs index bea14b936..825d4b7ee 100644 --- a/awa-model/src/insert.rs +++ b/awa-model/src/insert.rs @@ -23,7 +23,12 @@ use sqlx::{PgExecutor, PgPool}; // payloads) and then registers and reserves its lanes one at a time in // 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. +// lock it is held until the transaction ends. While it holds claims or lanes +// it waits at most 500 ms (or half of its session's `deadlock_timeout`, if +// lower) for another; a longer wait releases them all and retries, so a +// per-row producer that holds a lane and waits for one of the batch's keys +// usually proceeds instead of deadlocking with it. After 2 s of retries it +// makes one last attempt with ordinary lock waits. // // 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 @@ -534,14 +539,16 @@ const INSERT_MANY_BATCH_SQL: &str = r#" /// TRUE when `awa.insert_jobs_compat_batch` exists with the body the /// `insert_jobs_compat_batch` schema patch installs, whoever may call it: the -/// patch's probe. The body check matches the per-lane registration inside the -/// reservation loop, so `awa migrate` replaces an earlier body that registered -/// every new lane before reserving any. Catalog lookup only. +/// patch's probe. The body check matches the end of the bounded-wait phase +/// and the per-wait caller lock_timeout, so `awa migrate` replaces a body that +/// waited for a lane without bound while holding its claims, or that +/// registered every new lane before reserving any. Catalog lookup only. pub(crate) const BATCH_FUNCTION_PATCH_APPLIED_SQL: &str = "SELECT EXISTS (\ SELECT 1 FROM pg_proc AS p WHERE p.oid = to_regprocedure(\ 'awa.insert_jobs_compat_batch(text[],text[],jsonb[],text[],smallint[],smallint[],\ timestamptz[],jsonb[],jsonb[],bytea[],text[],text[],boolean)') \ - AND p.prosrc ~ 'INSERT INTO queue_lanes \\(queue, priority\\)\\s+SELECT v_lane\\.queue')"; + AND p.prosrc ~ 'v_bounded_until := clock_timestamp\\(\\) \\+ 4 \\* v_wait_budget' \ + AND p.prosrc ~ 'v_caller_wait_ms BETWEEN 1 AND v_wait_ms')"; /// TRUE when the connected role may call `awa.insert_jobs_compat_batch`. const BATCH_FUNCTION_USABLE_SQL: &str = "SELECT COALESCE(has_function_privilege(to_regprocedure(\ diff --git a/awa/tests/fixtures/insert_jobs_compat_batch_unbounded_lane_wait.sql b/awa/tests/fixtures/insert_jobs_compat_batch_unbounded_lane_wait.sql new file mode 100644 index 000000000..c169112eb --- /dev/null +++ b/awa/tests/fixtures/insert_jobs_compat_batch_unbounded_lane_wait.sql @@ -0,0 +1,533 @@ +-- The 0.6.15 insert_jobs_compat_batch body, which waited for each lane lock +-- for as long as it took while holding the batch's unique claims. Fixture for +-- the schema patch probe. + +CREATE OR REPLACE FUNCTION awa.insert_jobs_compat_batch( + p_kind TEXT[], + p_queue TEXT[], + p_args JSONB[], + p_state TEXT[], + p_priority SMALLINT[], + p_max_attempts SMALLINT[], + p_run_at TIMESTAMPTZ[], + p_metadata JSONB[], + p_tags JSONB[], + p_unique_key BYTEA[], + p_unique_states TEXT[], + p_ordering_hash TEXT[], + p_skip_unique_conflicts BOOLEAN DEFAULT FALSE +) +RETURNS SETOF awa.jobs +AS $$ +DECLARE + v_count INT := COALESCE(cardinality(p_kind), 0); + v_schema TEXT; + v_old_search_path TEXT; + v_now TIMESTAMPTZ := clock_timestamp(); + v_job_ids BIGINT[]; + v_skipped BIGINT[] := ARRAY[]::BIGINT[]; + v_lane RECORD; + v_lane_queues TEXT[] := ARRAY[]::TEXT[]; + v_lane_priorities SMALLINT[] := ARRAY[]::SMALLINT[]; + v_lane_shards SMALLINT[] := ARRAY[]::SMALLINT[]; + v_lane_starts BIGINT[] := ARRAY[]::BIGINT[]; + v_start BIGINT; + v_ready_slot INT; + v_ready_generation BIGINT; + v_row RECORD; + v_shards SMALLINT[]; +BEGIN + IF v_count = 0 THEN + RETURN; + END IF; + + IF cardinality(p_queue) IS DISTINCT FROM v_count + OR cardinality(p_args) IS DISTINCT FROM v_count + OR cardinality(p_state) IS DISTINCT FROM v_count + OR cardinality(p_priority) IS DISTINCT FROM v_count + OR cardinality(p_max_attempts) IS DISTINCT FROM v_count + OR cardinality(p_run_at) IS DISTINCT FROM v_count + OR cardinality(p_metadata) IS DISTINCT FROM v_count + OR cardinality(p_tags) IS DISTINCT FROM v_count + OR cardinality(p_unique_key) IS DISTINCT FROM v_count + OR cardinality(p_unique_states) IS DISTINCT FROM v_count + OR cardinality(p_ordering_hash) IS DISTINCT FROM v_count + THEN + RAISE EXCEPTION 'insert_jobs_compat_batch: column arrays must have equal length' + USING ERRCODE = '22023'; + END IF; + + IF EXISTS ( + SELECT 1 + FROM unnest(p_kind, p_queue, p_args, p_state, p_priority, p_max_attempts, p_metadata, p_tags) + AS input(kind, queue, args, state, priority, max_attempts, metadata, tags) + WHERE input.kind IS NULL OR input.queue IS NULL OR input.args IS NULL + OR input.state IS NULL OR input.priority IS NULL OR input.max_attempts IS NULL + OR input.metadata IS NULL OR input.tags IS NULL + ) THEN + RAISE EXCEPTION 'insert_jobs_compat_batch: kind, queue, args, state, priority, max_attempts, metadata and tags must not be NULL' + USING ERRCODE = '22004'; + END IF; + + IF EXISTS (SELECT 1 FROM unnest(p_kind) AS kind WHERE length(kind) > 200) THEN + RAISE EXCEPTION 'job kind length must be <= 200 characters' + USING ERRCODE = '23514'; + END IF; + + IF EXISTS (SELECT 1 FROM unnest(p_queue) AS queue WHERE length(queue) > 200) THEN + RAISE EXCEPTION 'queue name length must be <= 200 characters' + USING ERRCODE = '23514'; + END IF; + + IF EXISTS ( + SELECT 1 FROM unnest(p_priority) AS priority + WHERE priority < 1 OR priority > 4 + ) THEN + RAISE EXCEPTION 'priority must be between 1 and 4' + USING ERRCODE = '23514'; + END IF; + + IF EXISTS ( + SELECT 1 FROM unnest(p_max_attempts) AS max_attempts + WHERE max_attempts < 1 OR max_attempts > 1000 + ) THEN + RAISE EXCEPTION 'max_attempts must be between 1 and 1000' + USING ERRCODE = '23514'; + END IF; + + IF EXISTS ( + SELECT 1 FROM unnest(p_tags) AS tags + WHERE jsonb_array_length(tags) > 20 + ) THEN + RAISE EXCEPTION 'job tags must contain at most 20 values' + USING ERRCODE = '23514'; + END IF; + + v_schema := awa.active_queue_storage_schema(); + + -- Canonical storage has no lane lock to shorten: insert row by row. + IF v_schema IS NULL THEN + IF NOT p_skip_unique_conflicts THEN + RETURN QUERY + SELECT inserted.* + FROM unnest( + p_kind, p_queue, p_args, p_state, p_priority, p_max_attempts, + p_run_at, p_metadata, p_tags, p_unique_key, p_unique_states + ) WITH ORDINALITY AS input( + kind, queue, args, state, priority, max_attempts, run_at, + metadata, tags, unique_key, unique_states, ord + ) + CROSS JOIN LATERAL awa.insert_job_compat( + input.kind, + input.queue, + input.args, + input.state::awa.job_state, + input.priority, + input.max_attempts, + input.run_at, + input.metadata, + ARRAY( + SELECT tag.value + FROM jsonb_array_elements_text(input.tags) WITH ORDINALITY AS tag(value, tag_ord) + ORDER BY tag.tag_ord + ), + input.unique_key, + input.unique_states::bit(8), + NULL + ) AS inserted + ORDER BY input.ord; + RETURN; + END IF; + + -- Skipping needs a subtransaction per row. + FOR v_row IN + SELECT + input.ord, + input.kind, + input.queue, + input.args, + input.state, + input.priority, + input.max_attempts, + input.run_at, + input.metadata, + ARRAY( + SELECT tag.value + FROM jsonb_array_elements_text(input.tags) WITH ORDINALITY AS tag(value, tag_ord) + ORDER BY tag.tag_ord + ) AS tags, + input.unique_key, + input.unique_states + FROM unnest( + p_kind, p_queue, p_args, p_state, p_priority, p_max_attempts, + p_run_at, p_metadata, p_tags, p_unique_key, p_unique_states + ) WITH ORDINALITY AS input( + kind, queue, args, state, priority, max_attempts, run_at, + metadata, tags, unique_key, unique_states, ord + ) + ORDER BY input.ord + LOOP + BEGIN + RETURN NEXT awa.insert_job_compat( + v_row.kind, + v_row.queue, + v_row.args, + v_row.state::awa.job_state, + v_row.priority, + v_row.max_attempts, + v_row.run_at, + v_row.metadata, + v_row.tags, + v_row.unique_key, + v_row.unique_states::bit(8), + NULL + ); + EXCEPTION WHEN unique_violation THEN + NULL; + END; + END LOOP; + RETURN; + END IF; + + IF EXISTS ( + SELECT 1 FROM unnest(p_state) AS state + WHERE state NOT IN ('available', 'scheduled', 'retryable') + ) THEN + RAISE EXCEPTION 'queue storage does not support initial state %', + (SELECT state FROM unnest(p_state) AS state + WHERE state NOT IN ('available', 'scheduled', 'retryable') LIMIT 1) + USING ERRCODE = '22023'; + END IF; + + v_old_search_path := current_setting('search_path'); + PERFORM set_config('search_path', format('%I,awa,public', v_schema), true); + + -- Everything up to the lane reservation is lane-independent and runs + -- without a lane lock. + EXECUTE format( + 'SELECT array_agg(nextval(%L::regclass)::bigint) FROM generate_series(1, $1)', + format('%I.job_id_seq', v_schema) + ) + INTO v_job_ids + USING v_count; + + -- Rows that enqueue as `available` go to a lane. Their shard is derived as + -- insert_job_compat derives it: ordering-key rows from the key hash, the + -- rest from the backend pid. + SELECT array_agg( + CASE WHEN input.state = 'available' THEN + MOD( + CASE WHEN input.ordering_hash IS NULL + THEN pg_backend_pid()::numeric + ELSE input.ordering_hash::numeric + END, + COALESCE(meta.enqueue_shards, 1)::numeric + )::smallint + END + ORDER BY input.ord + ) + INTO v_shards + FROM unnest(p_queue, p_state, p_ordering_hash) + WITH ORDINALITY AS input(queue, state, ordering_hash, ord) + LEFT JOIN awa.queue_meta AS meta ON meta.queue = input.queue; + + -- Claims go in unique-key order, so two producers with overlapping keys + -- cannot wait on each other in opposite orders. + IF p_skip_unique_conflicts THEN + WITH claimed AS ( + INSERT INTO awa.job_unique_claims (unique_key, job_id) + SELECT input.unique_key, v_job_ids[input.ord] + FROM unnest(p_unique_key, p_unique_states, p_state) + WITH ORDINALITY AS input(unique_key, unique_states, state, ord) + WHERE input.unique_key IS NOT NULL + AND input.unique_states IS NOT NULL + AND awa.job_state_in_bitmask(input.unique_states::bit(8), input.state::awa.job_state) + ORDER BY input.unique_key, input.ord + ON CONFLICT (unique_key) DO NOTHING + RETURNING job_id + ) + SELECT COALESCE(array_agg(wanted.job_id), ARRAY[]::BIGINT[]) + INTO v_skipped + FROM ( + SELECT v_job_ids[input.ord] AS job_id + FROM unnest(p_unique_key, p_unique_states, p_state) + WITH ORDINALITY AS input(unique_key, unique_states, state, ord) + WHERE input.unique_key IS NOT NULL + AND input.unique_states IS NOT NULL + AND awa.job_state_in_bitmask(input.unique_states::bit(8), input.state::awa.job_state) + ) AS wanted + WHERE NOT EXISTS (SELECT 1 FROM claimed WHERE claimed.job_id = wanted.job_id); + ELSE + INSERT INTO awa.job_unique_claims (unique_key, job_id) + SELECT input.unique_key, v_job_ids[input.ord] + FROM unnest(p_unique_key, p_unique_states, p_state) + WITH ORDINALITY AS input(unique_key, unique_states, state, ord) + WHERE input.unique_key IS NOT NULL + AND input.unique_states IS NOT NULL + AND awa.job_state_in_bitmask(input.unique_states::bit(8), input.state::awa.job_state) + ORDER BY input.unique_key, input.ord; + END IF; + + -- Re-enqueues that are not yet due carry no lane. + INSERT INTO deferred_jobs ( + job_id, kind, queue, args, state, priority, attempt, run_lease, + max_attempts, run_at, attempted_at, finalized_at, created_at, + unique_key, unique_states, payload + ) + SELECT + v_job_ids[input.ord], + input.kind, + input.queue, + input.args, + input.state::awa.job_state, + input.priority, + 0, + 0, + input.max_attempts, + COALESCE(input.run_at, v_now), + NULL, + NULL, + v_now, + input.unique_key, + input.unique_states, + jsonb_build_object( + 'metadata', input.metadata, + 'tags', input.tags, + 'errors', '[]'::jsonb, + 'progress', NULL + ) + FROM unnest( + p_kind, p_queue, p_args, p_state, p_priority, p_max_attempts, p_run_at, + p_metadata, p_tags, p_unique_key, p_unique_states + ) WITH ORDINALITY AS input( + kind, queue, args, state, priority, max_attempts, run_at, + metadata, tags, unique_key, unique_states, ord + ) + WHERE input.state <> 'available' + AND v_job_ids[input.ord] <> ALL (v_skipped); + + -- The result rows do not depend on the lanes, so build them before the + -- lock window opens. + RETURN QUERY + SELECT + v_job_ids[input.ord], + input.kind, + input.queue, + input.args, + input.state::awa.job_state, + input.priority, + 0::smallint, + input.max_attempts, + COALESCE(input.run_at, v_now), + NULL::timestamptz, + NULL::timestamptz, + NULL::timestamptz, + NULL::timestamptz, + v_now, + ARRAY[]::jsonb[], + input.metadata, + ARRAY( + SELECT tag.value + FROM jsonb_array_elements_text(input.tags) WITH ORDINALITY AS tag(value, tag_ord) + ORDER BY tag.tag_ord + ), + input.unique_key, + input.unique_states::bit(8), + NULL::uuid, + NULL::timestamptz, + NULL::text, + NULL::text, + NULL::text, + NULL::text, + 0::bigint, + NULL::jsonb + FROM unnest( + p_kind, p_queue, p_args, p_state, p_priority, p_max_attempts, p_run_at, + p_metadata, p_tags, p_unique_key, p_unique_states + ) WITH ORDINALITY AS input( + kind, queue, args, state, priority, max_attempts, run_at, + metadata, tags, unique_key, unique_states, ord + ) + WHERE v_job_ids[input.ord] <> ALL (v_skipped) + ORDER BY input.ord; + + -- Lane lock window: in (queue, priority, shard) order, register each lane + -- the batch is the first to use and reserve its sequence range once, then + -- write the ready rows. The locks are held to the end of the transaction, + -- which keeps the rows of a lane visible in lane_seq order. + -- + -- A new lane is registered only when the loop reaches it, as every other + -- producer does: the head tables' row trigger provisions the lane under + -- its lock, and the uncommitted head row blocks other producers of that + -- lane, so registering a later lane before reserving an earlier one would + -- hold it out of order. Only missing rows are proposed, because the + -- trigger fires before the conflict check. + FOR v_lane IN + SELECT lanes.queue, lanes.priority, lanes.shard, lanes.row_count + FROM ( + SELECT input.queue, input.priority, v_shards[input.ord] AS shard, + count(*) AS row_count + FROM unnest(p_queue, p_priority, p_state) + WITH ORDINALITY AS input(queue, priority, state, ord) + WHERE input.state = 'available' + AND v_job_ids[input.ord] <> ALL (v_skipped) + GROUP BY input.queue, input.priority, v_shards[input.ord] + ) AS lanes + ORDER BY lanes.queue COLLATE "C", lanes.priority, lanes.shard + LOOP + INSERT INTO queue_lanes (queue, priority) + SELECT v_lane.queue, v_lane.priority + WHERE NOT EXISTS ( + SELECT 1 FROM queue_lanes AS existing + WHERE existing.queue = v_lane.queue AND existing.priority = v_lane.priority + ) + ON CONFLICT ON CONSTRAINT queue_lanes_pkey DO NOTHING; + + INSERT INTO queue_enqueue_heads (queue, priority, enqueue_shard) + SELECT v_lane.queue, v_lane.priority, v_lane.shard + WHERE NOT EXISTS ( + SELECT 1 FROM queue_enqueue_heads AS existing + WHERE existing.queue = v_lane.queue + AND existing.priority = v_lane.priority + AND existing.enqueue_shard = v_lane.shard + ) + ON CONFLICT (queue, priority, enqueue_shard) DO NOTHING; + + INSERT INTO queue_claim_heads (queue, priority, enqueue_shard) + SELECT v_lane.queue, v_lane.priority, v_lane.shard + WHERE NOT EXISTS ( + SELECT 1 FROM queue_claim_heads AS existing + WHERE existing.queue = v_lane.queue + AND existing.priority = v_lane.priority + AND existing.enqueue_shard = v_lane.shard + ) + ON CONFLICT (queue, priority, enqueue_shard) DO NOTHING; + + EXECUTE format('SELECT %I.reserve_enqueue_seq($1, $2, $3, $4)', v_schema) + INTO v_start + USING v_lane.queue, v_lane.priority, v_lane.shard, v_lane.row_count::BIGINT; + + v_lane_queues := v_lane_queues || v_lane.queue; + v_lane_priorities := v_lane_priorities || v_lane.priority; + v_lane_shards := v_lane_shards || v_lane.shard; + v_lane_starts := v_lane_starts || v_start; + END LOOP; + + IF cardinality(v_lane_queues) > 0 THEN + IF to_regprocedure(format('%I.ring_cursor(text)', v_schema)) IS NOT NULL THEN + EXECUTE format('SELECT slot, generation FROM %I.ring_cursor(''queue'')', v_schema) + INTO v_ready_slot, v_ready_generation; + ELSE + SELECT ring.current_slot, ring.generation + INTO v_ready_slot, v_ready_generation + FROM queue_ring_state AS ring + WHERE ring.singleton = TRUE; + END IF; + + INSERT INTO ready_entries ( + ready_slot, ready_generation, job_id, kind, queue, args, priority, + attempt, run_lease, max_attempts, lane_seq, enqueue_shard, run_at, + attempted_at, created_at, unique_key, unique_states, payload + ) + SELECT + v_ready_slot, + v_ready_generation, + v_job_ids[placed.ord], + placed.kind, + placed.queue, + placed.args, + placed.priority, + 0, + 0, + placed.max_attempts, + lane.start_seq + placed.lane_rank - 1, + lane.shard, + COALESCE(placed.run_at, v_now), + NULL, + v_now, + placed.unique_key, + placed.unique_states, + jsonb_build_object( + 'metadata', placed.metadata, + 'tags', placed.tags, + 'errors', '[]'::jsonb, + 'progress', NULL + ) + FROM ( + SELECT + input.*, + v_shards[input.ord] AS shard, + row_number() OVER ( + PARTITION BY input.queue, input.priority, v_shards[input.ord] + ORDER BY input.ord + ) AS lane_rank + FROM unnest( + p_kind, p_queue, p_args, p_state, p_priority, p_max_attempts, p_run_at, + p_metadata, p_tags, p_unique_key, p_unique_states + ) WITH ORDINALITY AS input( + kind, queue, args, state, priority, max_attempts, run_at, + metadata, tags, unique_key, unique_states, ord + ) + WHERE input.state = 'available' + AND v_job_ids[input.ord] <> ALL (v_skipped) + ) AS placed + JOIN unnest(v_lane_queues, v_lane_priorities, v_lane_shards, v_lane_starts) + AS lane(queue, priority, shard, start_seq) + ON lane.queue = placed.queue + AND lane.priority = placed.priority + AND lane.shard = placed.shard; + + -- One segment per run of consecutive lane_seq values sharing a run_at. + INSERT INTO ready_segments ( + ready_slot, ready_generation, queue, priority, enqueue_shard, + first_lane_seq, next_lane_seq, first_run_at + ) + SELECT + v_ready_slot, + v_ready_generation, + runs.queue, + runs.priority, + runs.shard, + min(runs.lane_seq), + max(runs.lane_seq) + 1, + runs.run_at + FROM ( + SELECT + placed.*, + placed.lane_seq - row_number() OVER ( + PARTITION BY placed.queue, placed.priority, placed.shard, placed.run_at + ORDER BY placed.lane_seq + ) AS run_id + FROM ( + SELECT + input.queue, + input.priority, + v_shards[input.ord] AS shard, + COALESCE(input.run_at, v_now) AS run_at, + lane.start_seq + row_number() OVER ( + PARTITION BY input.queue, input.priority, v_shards[input.ord] + ORDER BY input.ord + ) - 1 AS lane_seq + FROM unnest(p_queue, p_priority, p_state, p_run_at) + WITH ORDINALITY AS input(queue, priority, state, run_at, ord) + JOIN unnest(v_lane_queues, v_lane_priorities, v_lane_shards, v_lane_starts) + AS lane(queue, priority, shard, start_seq) + ON lane.queue = input.queue + AND lane.priority = input.priority + AND lane.shard = v_shards[input.ord] + WHERE input.state = 'available' + AND v_job_ids[input.ord] <> ALL (v_skipped) + ) AS placed + ) AS runs + GROUP BY runs.queue, runs.priority, runs.shard, runs.run_at, runs.run_id + ON CONFLICT (ready_slot, ready_generation, queue, priority, enqueue_shard, first_lane_seq) + DO NOTHING; + + PERFORM pg_notify('awa:' || notified.queue, '') + FROM (SELECT DISTINCT unnest(v_lane_queues) AS queue) AS notified; + END IF; + + PERFORM set_config('search_path', v_old_search_path, true); +END; +$$ LANGUAGE plpgsql VOLATILE +SET search_path = pg_catalog, awa, public; diff --git a/awa/tests/insert_many_batch_test.rs b/awa/tests/insert_many_batch_test.rs index b7446cd12..cbd428f8b 100644 --- a/awa/tests/insert_many_batch_test.rs +++ b/awa/tests/insert_many_batch_test.rs @@ -17,7 +17,9 @@ use awa::model::{ }; use awa::{InsertOpts, InsertParams, JobState, QueueStorageConfig, UniqueOpts}; use awa_testing::setup; +use sqlx::postgres::{PgConnectOptions, PgPoolOptions}; use sqlx::PgPool; +use std::str::FromStr; use std::sync::LazyLock; use std::time::{Duration, Instant}; use tokio::sync::Mutex; @@ -113,13 +115,22 @@ enum Path { impl Path { const ALL: [Path; 3] = [Path::PerRow, Path::SkipUniqueConflicts, Path::Copy]; - async fn insert(self, pool: &PgPool, jobs: &[InsertParams]) -> Vec { + async fn try_insert( + self, + pool: &PgPool, + jobs: &[InsertParams], + ) -> Result, AwaError> { match self { Path::PerRow => insert_many(pool, jobs).await, Path::SkipUniqueConflicts => insert_many_skip_unique_conflicts(pool, jobs).await, Path::Copy => insert_many_copy_from_pool(pool, jobs).await, } - .unwrap_or_else(|err| panic!("{self:?} insert: {err:?}")) + } + + async fn insert(self, pool: &PgPool, jobs: &[InsertParams]) -> Vec { + self.try_insert(pool, jobs) + .await + .unwrap_or_else(|err| panic!("{self:?} insert: {err:?}")) } /// Ready segments a batch of `rows` ready rows in one lane, sharing a @@ -533,14 +544,28 @@ async fn batch_function_without_skip_fails_the_batch_on_a_unique_conflict() { async fn batch_waits_for_the_lane_after_its_lane_independent_work() { let _guard = ENGINE_GUARD.lock().await; let pool = queue_storage_pool().await; + // The batch keeps its claims only for its first attempt's wait budget + // (500 ms by default), so a probe that lands later shows nothing; such a + // run is repeated rather than judged. + for _ in 0..3 { + if lane_window_holds_claims(&pool).await { + return; + } + } + panic!("every probe of the batch's claims landed after its first attempt's wait budget"); +} + +/// Returns false when the probe landed too late to judge. +async fn lane_window_holds_claims(pool: &PgPool) -> bool { + const JUDGEABLE: Duration = Duration::from_millis(400); let queue = unique_queue("window"); // Provision the lane so the gate lock is the only thing to wait for. - insert_many(&pool, &[job(&queue, -1, false)]) + insert_many(pool, &[job(&queue, -1, false)]) .await .expect("provision lane"); - let key = lane_lock_key(&pool, &queue).await; + let key = lane_lock_key(pool, &queue).await; let mut gate = pool.begin().await.expect("begin gate"); sqlx::query("SELECT pg_advisory_xact_lock($1)") .bind(key) @@ -549,13 +574,13 @@ async fn batch_waits_for_the_lane_after_its_lane_independent_work() { .expect("take lane lock"); let jobs_for_probe: Vec = (0..50).map(|seq| job(&queue, seq, true)).collect(); + let started = Instant::now(); let producer = { let pool = pool.clone(); let jobs = jobs_for_probe.clone(); tokio::spawn(async move { insert_many_skip_unique_conflicts(&pool, &jobs).await }) }; - let started = Instant::now(); let waiter = loop { let pid: Option = sqlx::query_scalar( "SELECT pid FROM pg_locks @@ -564,33 +589,16 @@ async fn batch_waits_for_the_lane_after_its_lane_independent_work() { LIMIT 1", ) .bind(key) - .fetch_optional(&pool) + .fetch_optional(pool) .await .expect("poll pg_locks"); if let Some(pid) = pid { break pid; } assert!(started.elapsed() < WAIT_DEADLINE, "producer never waited"); - tokio::time::sleep(Duration::from_millis(10)).await; + tokio::time::sleep(Duration::from_millis(5)).await; }; - let holds_relation_lock = |relation: &'static str| { - let pool = pool.clone(); - async move { - sqlx::query_scalar::<_, bool>( - "SELECT EXISTS ( - SELECT 1 FROM pg_locks - WHERE pid = $1 AND locktype = 'relation' AND granted - AND relation = $2::regclass - )", - ) - .bind(waiter) - .bind(relation) - .fetch_one(&pool) - .await - .expect("read relation locks") - } - }; // The claim of the batch's last row is already held, uncommitted: an // insert of the same key waits on the producer's transaction. let last = jobs_for_probe.last().expect("probe job"); @@ -601,7 +609,7 @@ async fn batch_waits_for_the_lane_after_its_lane_independent_work() { .expect("unique key") .to_vec(); let mut probe = pool.begin().await.expect("begin probe"); - sqlx::query("SET LOCAL lock_timeout = '300ms'") + sqlx::query("SET LOCAL lock_timeout = '20ms'") .execute(&mut *probe) .await .expect("set lock timeout"); @@ -612,17 +620,38 @@ async fn batch_waits_for_the_lane_after_its_lane_independent_work() { .bind(&last_key) .execute(&mut *probe) .await; + let probed_after = started.elapsed(); match probe_result { - Err(sqlx::Error::Database(db_err)) => assert_eq!( - db_err.code().as_deref(), - Some("55P03"), - "the last row's claim must already be held while the lane lock is awaited" + Err(sqlx::Error::Database(db_err)) if db_err.code().as_deref() == Some("55P03") => {} + Ok(_) if probed_after > JUDGEABLE => { + probe.rollback().await.expect("roll back probe"); + gate.rollback().await.expect("release lane lock"); + tokio::time::timeout(WAIT_DEADLINE, producer) + .await + .expect("producer finished") + .expect("producer task") + .expect("insert batch"); + return false; + } + other => panic!( + "claim of the last row was not held while waiting for the lane \ + ({probed_after:?} after the batch started): {other:?}" ), - other => panic!("claim of the last row was not held while waiting for the lane: {other:?}"), } drop(probe); + let writes_ready_rows: bool = sqlx::query_scalar( + "SELECT EXISTS ( + SELECT 1 FROM pg_locks + WHERE pid = $1 AND locktype = 'relation' AND granted + AND relation = 'awa.ready_entries'::regclass + )", + ) + .bind(waiter) + .fetch_one(pool) + .await + .expect("read relation locks"); assert!( - !holds_relation_lock("awa.ready_entries").await, + !writes_ready_rows, "no ready row is written while the lane lock is awaited" ); @@ -633,7 +662,8 @@ async fn batch_waits_for_the_lane_after_its_lane_independent_work() { .expect("producer task") .expect("insert batch"); assert_eq!(inserted.len(), 50); - assert_eq!(ready_lane_seqs(&pool, &queue).await.len(), 51); + assert_eq!(ready_lane_seqs(pool, &queue).await.len(), 51); + true } /// Producers on the per-row function, as released 0.6 binaries are, and @@ -804,3 +834,733 @@ async fn assert_batch_does_not_hold_a_new_lane_while_waiting(pool: &PgPool, path assert_eq!(ready_lane_seqs(pool, &lane_a).await.len(), 3, "{path:?}"); assert_eq!(ready_lane_seqs(pool, &lane_b).await.len(), 2, "{path:?}"); } + +/// Polls until some session waits for the lane lock `key`. +async fn wait_until_lane_awaited(pool: &PgPool, key: i64, what: &str) { + let started = Instant::now(); + loop { + let waiting: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM pg_locks + WHERE locktype = 'advisory' AND NOT granted + AND ((classid::bigint << 32) | objid::bigint) = $1)", + ) + .bind(key) + .fetch_one(pool) + .await + .expect("poll pg_locks"); + if waiting { + return; + } + assert!(started.elapsed() < WAIT_DEADLINE, "{what} never waited"); + tokio::time::sleep(Duration::from_millis(10)).await; + } +} + +/// A per-row producer holds the lane and then inserts a unique key that a +/// waiting batch has claimed but not committed. The batch gives up its claims +/// within its bounded lane wait, so the per-row insert proceeds instead of +/// both transactions waiting on each other until PostgreSQL aborts one with +/// `deadlock detected`. The batch then finds the key claimed and skips it. +#[tokio::test] +async fn per_row_insert_of_a_key_a_waiting_batch_claimed_does_not_deadlock() { + let _guard = ENGINE_GUARD.lock().await; + let pool = queue_storage_pool().await; + for path in [Path::SkipUniqueConflicts, Path::Copy] { + assert_batch_gives_up_claims_while_waiting_for_a_lane(&pool, &pool, path).await; + } +} + +/// The same cycle with the batch's session at a `deadlock_timeout` far above +/// the per-row producer's default of one second. The batch's wait budget is +/// capped, so it still gives way before the per-row producer's deadlock check. +#[tokio::test] +async fn batch_with_a_long_deadlock_timeout_still_gives_way_to_a_per_row_producer() { + let _guard = ENGINE_GUARD.lock().await; + let pool = queue_storage_pool().await; + let batch_pool = PgPoolOptions::new() + .max_connections(2) + .connect_with( + PgConnectOptions::from_str(&setup::database_url()) + .expect("parse database url") + .options([("deadlock_timeout", "20s")]), + ) + .await + .expect("connect with a long deadlock_timeout"); + for path in [Path::SkipUniqueConflicts, Path::Copy] { + assert_batch_gives_up_claims_while_waiting_for_a_lane(&pool, &batch_pool, path).await; + } +} + +async fn assert_batch_gives_up_claims_while_waiting_for_a_lane( + pool: &PgPool, + batch_pool: &PgPool, + path: Path, +) { + let queue = unique_queue("claim_cycle"); + insert_many(pool, &[job(&queue, -1, false)]) + .await + .expect("provision lane"); + + let mut per_row = pool.begin().await.expect("begin per-row producer"); + insert_many(&mut *per_row, &[job(&queue, 100, false)]) + .await + .expect("per-row insert takes the lane"); + + let batch = { + let batch_pool = batch_pool.clone(); + let jobs = vec![job(&queue, 0, true)]; + tokio::spawn(async move { path.try_insert(&batch_pool, &jobs).await }) + }; + wait_until_lane_awaited(pool, lane_lock_key(pool, &queue).await, "batch").await; + + let per_row_insert = tokio::time::timeout( + WAIT_DEADLINE, + insert_many(&mut *per_row, &[job(&queue, 0, true)]), + ) + .await + .unwrap_or_else(|_| panic!("{path:?}: per-row insert of the claimed key never finished")); + let inserted_by_row = per_row_insert + .unwrap_or_else(|err| panic!("{path:?}: per-row insert of the claimed key: {err:?}")); + assert_eq!(inserted_by_row.len(), 1, "{path:?}"); + per_row.commit().await.expect("commit per-row producer"); + + let inserted_by_batch = tokio::time::timeout(WAIT_DEADLINE, batch) + .await + .expect("batch finished") + .expect("batch task") + .unwrap_or_else(|err| panic!("{path:?}: batch: {err:?}")); + assert!( + inserted_by_batch.is_empty(), + "{path:?}: the batch skips the key the per-row producer committed" + ); + let seqs: Vec = ready_lane_seqs(pool, &queue) + .await + .into_iter() + .map(|(seq, _)| seq) + .collect(); + assert_eq!(seqs, vec![-1, 100, 0], "{path:?}"); + assert_eq!(claims_for_queue(pool, &queue).await, 1, "{path:?}"); +} + +/// A transaction that enqueues lane B and then lane A in two statements, and +/// a batch for lanes A and B that holds A while it waits for B. The batch +/// releases A within its bounded wait instead of deadlocking, then waits for +/// B without holding anything and lands once the transaction commits. +#[tokio::test] +async fn multi_statement_producer_and_batch_in_opposite_lane_order_do_not_deadlock() { + let _guard = ENGINE_GUARD.lock().await; + let pool = queue_storage_pool().await; + for path in [Path::SkipUniqueConflicts, Path::Copy] { + let suffix = &Uuid::new_v4().simple().to_string()[..12]; + let lane_a = format!("batch_order_a_{suffix}"); + let lane_b = format!("batch_order_b_{suffix}"); + insert_many(&pool, &[job(&lane_a, -1, false), job(&lane_b, -1, false)]) + .await + .expect("provision lanes"); + + let mut producer = pool.begin().await.expect("begin producer"); + insert_many(&mut *producer, &[job(&lane_b, 100, false)]) + .await + .expect("producer takes lane B"); + + let batch = { + let pool = pool.clone(); + let jobs = vec![job(&lane_a, 0, false), job(&lane_b, 0, false)]; + tokio::spawn(async move { path.try_insert(&pool, &jobs).await }) + }; + wait_until_lane_awaited(&pool, lane_lock_key(&pool, &lane_b).await, "batch").await; + + tokio::time::timeout( + WAIT_DEADLINE, + insert_many(&mut *producer, &[job(&lane_a, 100, false)]), + ) + .await + .unwrap_or_else(|_| panic!("{path:?}: producer never got lane A")) + .unwrap_or_else(|err| panic!("{path:?}: producer insert into lane A: {err:?}")); + producer.commit().await.expect("commit producer"); + + let inserted = tokio::time::timeout(WAIT_DEADLINE, batch) + .await + .expect("batch finished") + .expect("batch task") + .unwrap_or_else(|err| panic!("{path:?}: batch: {err:?}")); + assert_eq!(inserted.len(), 2, "{path:?}"); + let lane_a_seqs: Vec = ready_lane_seqs(&pool, &lane_a) + .await + .into_iter() + .map(|(seq, _)| seq) + .collect(); + assert_eq!(lane_a_seqs, vec![-1, 100, 0], "{path:?}"); + let lane_b_seqs: Vec = ready_lane_seqs(&pool, &lane_b) + .await + .into_iter() + .map(|(seq, _)| seq) + .collect(); + assert_eq!(lane_b_seqs, vec![-1, 100, 0], "{path:?}"); + } +} + +/// The caller's `lock_timeout` still ends a batch's lock wait: a lane held +/// throughout fails it with `lock_not_available`, as it fails a per-row +/// insert, and the batch leaves its unique claims behind. +#[tokio::test] +async fn batch_gives_up_at_the_callers_lock_timeout() { + let _guard = ENGINE_GUARD.lock().await; + let pool = queue_storage_pool().await; + let queue = unique_queue("caller_timeout"); + insert_many(&pool, &[job(&queue, -1, false)]) + .await + .expect("provision lane"); + + let mut gate = pool.begin().await.expect("begin gate"); + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(lane_lock_key(&pool, &queue).await) + .execute(&mut *gate) + .await + .expect("take lane lock"); + + let mut producer = pool.begin().await.expect("begin producer"); + sqlx::query("SET LOCAL lock_timeout = '300ms'") + .execute(&mut *producer) + .await + .expect("set lock timeout"); + let started = Instant::now(); + let err = insert_many_skip_unique_conflicts(&mut *producer, &[job(&queue, 0, true)]) + .await + .expect_err("the lane stays held past the caller's lock_timeout"); + assert!( + matches!(&err, AwaError::Database(sqlx::Error::Database(db_err)) + if db_err.code().as_deref() == Some("55P03")), + "{err:?}" + ); + assert!(started.elapsed() < WAIT_DEADLINE); + producer.rollback().await.expect("roll back producer"); + gate.rollback().await.expect("release lane lock"); + + let inserted = insert_many_skip_unique_conflicts(&pool, &[job(&queue, 0, true)]) + .await + .expect("insert once the lane is free"); + assert_eq!(inserted.len(), 1, "the key was not left claimed"); +} + +/// Batch and per-row producers race on the same unique keys and lanes, round +/// after round. Per-row producers enqueue the last lane in one statement and +/// the others in a second, so they hold a lane while they claim keys and wait +/// for lanes that batches hold. No producer may fail with `deadlock +/// detected`, every key lands exactly once, and the rows in the lanes are +/// exactly the rows the committed producers returned. +#[tokio::test] +async fn mixed_producers_on_shared_keys_and_lanes_never_deadlock() { + const TASKS: usize = 8; + const ROUNDS: usize = 12; + const KEYS: usize = 6; + + let _guard = ENGINE_GUARD.lock().await; + let pool = setup::setup(TASKS as u32 + 4).await; + setup::activate_queue_storage(&pool).await; + let suffix = Uuid::new_v4().simple().to_string()[..12].to_string(); + let lanes: Vec = ["a", "b", "c"] + .iter() + .map(|lane| format!("batch_mix_{lane}_{suffix}")) + .collect(); + insert_many( + &pool, + &lanes + .iter() + .map(|lane| job(lane, -1, false)) + .collect::>(), + ) + .await + .expect("provision lanes"); + + let barrier = std::sync::Arc::new(tokio::sync::Barrier::new(TASKS)); + let mut tasks = Vec::new(); + for task in 0..TASKS { + let pool = pool.clone(); + let barrier = barrier.clone(); + let lanes = lanes.clone(); + tasks.push(tokio::spawn(async move { + let mut committed: Vec = Vec::new(); + let mut deadlocks = 0usize; + for round in 0..ROUNDS { + // Each task takes the round's keys in its own order. + let mut keys: Vec = (0..KEYS).collect(); + keys.rotate_left((task * 5 + round) % KEYS); + if (task + round) % 2 == 1 { + keys.reverse(); + } + let mut jobs: Vec = keys + .iter() + .map(|&key| keyed_job(&lanes, round, key)) + .collect(); + jobs.extend((0..lanes.len()).map(|lane| filler_job(&lanes, round, task, lane))); + barrier.wait().await; + + let outcome: Result, AwaError> = match task % 4 { + 0 | 1 => { + // Per-row producers share one statement order, last + // lane first and keys ascending, so they cannot + // deadlock with each other. + let last_lane = &lanes[lanes.len() - 1]; + let mut sorted = jobs.clone(); + sorted.sort_by_key(|job| { + job.args + .get("key") + .and_then(|key| key.as_u64()) + .unwrap_or(u64::MAX) + }); + let (first, second): (Vec, Vec) = sorted + .into_iter() + .partition(|job| &job.opts.queue == last_lane); + async { + let mut tx = pool.begin().await?; + let mut rows = insert_many(&mut *tx, &first).await?; + rows.extend(insert_many(&mut *tx, &second).await?); + tx.commit().await?; + Ok(rows) + } + .await + } + 2 => insert_many_skip_unique_conflicts(&pool, &jobs).await, + _ => insert_many_copy_from_pool(&pool, &jobs).await, + }; + match outcome { + Ok(rows) => committed.extend(rows.iter().map(|row| row.id)), + Err(err) if err.is_deadlock() => deadlocks += 1, + Err(err) if task % 4 < 2 && is_unique_violation(&err) => {} + Err(err) => panic!("task {task} round {round}: {err:?}"), + } + } + (committed, deadlocks) + })); + } + + let mut committed: Vec = Vec::new(); + let mut deadlocks = 0usize; + for task in tasks { + let (ids, task_deadlocks) = task.await.expect("producer task"); + committed.extend(ids); + deadlocks += task_deadlocks; + } + assert_eq!(deadlocks, 0, "producers failed with deadlock detected"); + + let landed: Vec<(i64, serde_json::Value)> = sqlx::query_as( + "SELECT job_id, args FROM awa.ready_entries + WHERE queue = ANY($1) AND args ? 'round'", + ) + .bind(&lanes) + .fetch_all(&pool) + .await + .expect("read lanes"); + let mut landed_ids: Vec = landed.iter().map(|(id, _)| *id).collect(); + landed_ids.sort_unstable(); + committed.sort_unstable(); + assert_eq!( + landed_ids, committed, + "the lanes hold exactly the rows committed producers returned" + ); + for round in 0..ROUNDS { + for key in 0..KEYS { + let copies = landed + .iter() + .filter(|(_, args)| { + args.get("round") == Some(&serde_json::json!(round)) + && args.get("key") == Some(&serde_json::json!(key)) + }) + .count(); + assert_eq!(copies, 1, "round {round} key {key} lands exactly once"); + } + } +} + +/// Round `round`'s unique job `key`, in lane `key % lanes.len()`. +fn keyed_job(lanes: &[String], round: usize, key: usize) -> InsertParams { + InsertParams { + kind: "insert_many_batch_job".to_string(), + args: serde_json::json!({ "round": round, "key": key }), + opts: InsertOpts { + queue: lanes[key % lanes.len()].clone(), + unique: Some(UniqueOpts { + by_queue: true, + by_args: true, + ..Default::default() + }), + ..Default::default() + }, + } +} + +/// A job without a unique key that only `task` inserts. +fn filler_job(lanes: &[String], round: usize, task: usize, lane: usize) -> InsertParams { + InsertParams { + kind: "insert_many_batch_job".to_string(), + args: serde_json::json!({ "round": round, "task": task, "lane": lane }), + opts: InsertOpts { + queue: lanes[lane].clone(), + ..Default::default() + }, + } +} + +/// A batch whose caller already holds a lane from an earlier statement, and a +/// per-row producer that holds one of the batch's keys and then waits for that +/// lane. Rolling back the batch's attempts cannot release the caller's lane, +/// so this is a real deadlock. It must end, as it did before the batch had a +/// wait budget, with PostgreSQL aborting one side, not with the batch retrying +/// forever. The per-row producer's session has a long `deadlock_timeout`, so +/// only the batch's side can run the deadlock check in time. +#[tokio::test] +async fn batch_in_a_transaction_that_already_holds_a_lane_ends_a_cycle() { + let _guard = ENGINE_GUARD.lock().await; + let pool = queue_storage_pool().await; + let patient_pool = PgPoolOptions::new() + .max_connections(1) + .connect_with( + PgConnectOptions::from_str(&setup::database_url()) + .expect("parse database url") + .options([("deadlock_timeout", "60s")]), + ) + .await + .expect("connect with a long deadlock_timeout"); + let suffix = &Uuid::new_v4().simple().to_string()[..12]; + let lane_1 = format!("batch_caller_1_{suffix}"); + let lane_2 = format!("batch_caller_2_{suffix}"); + insert_many(&pool, &[job(&lane_1, -1, false), job(&lane_2, -1, false)]) + .await + .expect("provision lanes"); + + // The per-row producer claims the key and holds lane 2. + let mut per_row = patient_pool.begin().await.expect("begin per-row producer"); + insert_many(&mut *per_row, &[job(&lane_2, 0, true)]) + .await + .expect("per-row producer claims the key"); + + // The caller holds lane 1, then its batch waits for the key. + let (holds_lane_1, lane_1_held) = tokio::sync::oneshot::channel(); + let caller = { + let pool = pool.clone(); + let lane_1 = lane_1.clone(); + let lane_2 = lane_2.clone(); + tokio::spawn(async move { + let mut tx = pool.begin().await?; + insert_many(&mut *tx, &[job(&lane_1, 100, false)]).await?; + let _ = holds_lane_1.send(()); + let inserted = + insert_many_skip_unique_conflicts(&mut *tx, &[job(&lane_2, 0, true)]).await?; + tx.commit().await?; + Ok::<_, AwaError>(inserted) + }) + }; + lane_1_held.await.expect("the caller took lane 1"); + let per_row_outcome = tokio::time::timeout(WAIT_DEADLINE, async { + insert_many(&mut *per_row, &[job(&lane_1, 200, false)]).await?; + per_row.commit().await?; + Ok::<_, AwaError>(()) + }) + .await + .expect("the per-row producer's side of the cycle ended"); + let caller_outcome = tokio::time::timeout(WAIT_DEADLINE, caller) + .await + .expect("the caller's side of the cycle ended") + .expect("caller task"); + + let deadlocks = [ + per_row_outcome + .as_ref() + .err() + .is_some_and(AwaError::is_deadlock), + caller_outcome + .as_ref() + .err() + .is_some_and(AwaError::is_deadlock), + ]; + assert_eq!( + deadlocks.iter().filter(|aborted| **aborted).count(), + 1, + "exactly one side is aborted with deadlock detected: per-row {per_row_outcome:?}, \ + caller {caller_outcome:?}" + ); + assert!(per_row_outcome.is_ok() || caller_outcome.is_ok()); +} + +/// Two lanes held for longer than the batch's bounded phase: the batch's last +/// attempt waits for them as an ordinary producer would, and lands as soon as +/// they are released. +#[tokio::test] +async fn batch_lands_once_two_long_held_lanes_are_released() { + let _guard = ENGINE_GUARD.lock().await; + let pool = queue_storage_pool().await; + let suffix = &Uuid::new_v4().simple().to_string()[..12]; + let lane_a = format!("batch_held_a_{suffix}"); + let lane_b = format!("batch_held_b_{suffix}"); + insert_many(&pool, &[job(&lane_a, -1, false), job(&lane_b, -1, false)]) + .await + .expect("provision lanes"); + + let mut gates = Vec::new(); + for lane in [&lane_a, &lane_b] { + let mut gate = pool.begin().await.expect("begin gate"); + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(lane_lock_key(&pool, lane).await) + .execute(&mut *gate) + .await + .expect("take lane lock"); + gates.push(gate); + } + let started = Instant::now(); + let batch = { + let pool = pool.clone(); + let jobs = vec![job(&lane_a, 0, true), job(&lane_b, 0, true)]; + tokio::spawn(async move { insert_many_skip_unique_conflicts(&pool, &jobs).await }) + }; + // Past the bounded phase (four 500 ms budgets). + tokio::time::sleep(Duration::from_millis(3_000)).await; + assert!( + !batch.is_finished(), + "the batch waits while the lanes are held" + ); + for gate in gates { + gate.commit().await.expect("release lane"); + } + let inserted = tokio::time::timeout(WAIT_DEADLINE, batch) + .await + .expect("batch finished") + .expect("batch task") + .expect("insert batch"); + assert_eq!(inserted.len(), 2); + assert!( + started.elapsed() < Duration::from_secs(5), + "landed promptly on release" + ); + for lane in [&lane_a, &lane_b] { + let seqs: Vec = ready_lane_seqs(&pool, lane) + .await + .into_iter() + .map(|(seq, _)| seq) + .collect(); + assert_eq!(seqs, vec![-1, 0]); + } +} + +/// The claim cycle with a large batch whose last key in key order is the one +/// the per-row producer inserts. Preparing the result rows happens before the +/// wait budget starts, but inserting the claims counts against it: the batch +/// can only give way while its claims take less than the budget (500 ms), and +/// a larger batch behaves as in 0.6.15. The test first times inserting as +/// many claims. When that fits well inside the budget the batch must give way; otherwise the cycle must still end, with both sides done or +/// exactly one aborted by `deadlock detected`. +#[tokio::test] +async fn large_batch_gives_way_while_its_claims_fit_the_wait_budget() { + const ROWS: i64 = 5_000; + const FITS: Duration = Duration::from_millis(250); + let _guard = ENGINE_GUARD.lock().await; + let pool = queue_storage_pool().await; + + // What inserting this many claims costs here, the part of the batch's + // work that counts against its wait budget. + let mut calibration = pool.begin().await.expect("begin calibration"); + let started = Instant::now(); + sqlx::query( + "INSERT INTO awa.job_unique_claims (unique_key, job_id) + SELECT decode(md5($1 || g::text), 'hex'), -g + FROM generate_series(1, $2) AS g + ORDER BY 1 + ON CONFLICT (unique_key) DO NOTHING", + ) + .bind(unique_queue("large_calibration")) + .bind(ROWS) + .execute(&mut *calibration) + .await + .expect("time a claim insert"); + let claim_cost = started.elapsed(); + calibration.rollback().await.expect("roll back calibration"); + let fits_budget = claim_cost < FITS; + + let queue = unique_queue("large_cycle"); + insert_many(&pool, &[job(&queue, -1, false)]) + .await + .expect("provision lane"); + let jobs: Vec = (0..ROWS).map(|seq| job(&queue, seq, true)).collect(); + let last_in_key_order = jobs + .iter() + .max_by_key(|job| { + awa::model::prepare_raw_job_insert(job.kind.clone(), job.args.clone(), job.opts.clone()) + .expect("prepare job") + .unique_key() + .expect("unique key") + .to_vec() + }) + .expect("a job") + .clone(); + + let mut per_row = pool.begin().await.expect("begin per-row producer"); + insert_many(&mut *per_row, &[job(&queue, ROWS + 1, false)]) + .await + .expect("per-row insert takes the lane"); + let batch = { + let pool = pool.clone(); + tokio::spawn(async move { insert_many_skip_unique_conflicts(&pool, &jobs).await }) + }; + wait_until_lane_awaited(&pool, lane_lock_key(&pool, &queue).await, "batch").await; + + let per_row_outcome = tokio::time::timeout(WAIT_DEADLINE, async { + insert_many(&mut *per_row, &[last_in_key_order]).await?; + per_row.commit().await?; + Ok::<_, AwaError>(()) + }) + .await + .expect("the per-row producer's side of the cycle ended"); + let batch_outcome = tokio::time::timeout(WAIT_DEADLINE, batch) + .await + .expect("the batch's side of the cycle ended") + .expect("batch task"); + + if fits_budget { + per_row_outcome.unwrap_or_else(|err| { + panic!("per-row insert of the batch's last key (claims take {claim_cost:?}): {err:?}") + }); + let inserted = batch_outcome + .unwrap_or_else(|err| panic!("batch (claims take {claim_cost:?}): {err:?}")); + assert_eq!( + inserted.len() as i64, + ROWS - 1, + "the batch skips the committed key" + ); + assert_eq!(ready_lane_seqs(&pool, &queue).await.len() as i64, ROWS + 2); + } else { + let aborted = [ + &per_row_outcome.as_ref().err(), + &batch_outcome.as_ref().err(), + ] + .iter() + .filter(|err| err.is_some_and(AwaError::is_deadlock)) + .count(); + let both_done = per_row_outcome.is_ok() && batch_outcome.is_ok(); + assert!( + both_done || (aborted == 1 && (per_row_outcome.is_ok() || batch_outcome.is_ok())), + "a batch whose claims take {claim_cost:?} ends as in 0.6.15: per-row {per_row_outcome:?}, \ + batch {batch_outcome:?}" + ); + } +} + +/// The caller's `lock_timeout` applies to each lock wait, as it does in any +/// statement, not to the whole call: a batch whose own work before its lane +/// wait takes longer than the caller's `lock_timeout` still lands when the +/// lane is released within that timeout once it starts waiting. +#[tokio::test] +async fn callers_lock_timeout_applies_to_each_wait() { + const ROWS: i64 = 10_000; + let _guard = ENGINE_GUARD.lock().await; + let pool = queue_storage_pool().await; + let queue = unique_queue("caller_wait"); + insert_many(&pool, &[job(&queue, -1, false)]) + .await + .expect("provision lane"); + + let key = lane_lock_key(&pool, &queue).await; + let mut gate = pool.begin().await.expect("begin gate"); + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(key) + .execute(&mut *gate) + .await + .expect("take lane lock"); + + let jobs: Vec = (0..ROWS).map(|seq| job(&queue, seq, true)).collect(); + let batch = { + let pool = pool.clone(); + tokio::spawn(async move { + let mut tx = pool.begin().await?; + sqlx::query("SET LOCAL lock_timeout = '100ms'") + .execute(&mut *tx) + .await?; + let started = Instant::now(); + let inserted = insert_many_skip_unique_conflicts(&mut *tx, &jobs).await?; + let elapsed = started.elapsed(); + tx.commit().await?; + Ok::<_, AwaError>((inserted.len(), elapsed)) + }) + }; + let started = Instant::now(); + loop { + let waiting: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM pg_locks + WHERE locktype = 'advisory' AND NOT granted + AND ((classid::bigint << 32) | objid::bigint) = $1)", + ) + .bind(key) + .fetch_one(&pool) + .await + .expect("poll pg_locks"); + if waiting || batch.is_finished() { + break; + } + assert!(started.elapsed() < WAIT_DEADLINE, "batch never waited"); + tokio::time::sleep(Duration::from_millis(2)).await; + } + gate.commit().await.expect("release lane lock"); + + let (inserted, elapsed) = tokio::time::timeout(WAIT_DEADLINE, batch) + .await + .expect("batch finished") + .expect("batch task") + .expect("batch with a 100 ms lock_timeout"); + assert_eq!(inserted as i64, ROWS); + assert!( + elapsed > Duration::from_millis(100), + "the batch's own work outlasts the caller's lock_timeout ({elapsed:?})" + ); +} + +/// The documented limit of the bounded phase: a per-row producer that holds a +/// lane for longer than the batch's retries (about 2 s), and only then inserts +/// a key the batch has claimed, meets the batch's ordinary last attempt. The +/// cycle then ends as in 0.6.15, with one side aborted by `deadlock detected`. +#[tokio::test] +async fn per_row_producer_holding_a_lane_past_the_bounded_phase_meets_an_ordinary_wait() { + let _guard = ENGINE_GUARD.lock().await; + let pool = queue_storage_pool().await; + let queue = unique_queue("long_hold"); + insert_many(&pool, &[job(&queue, -1, false)]) + .await + .expect("provision lane"); + + let mut per_row = pool.begin().await.expect("begin per-row producer"); + insert_many(&mut *per_row, &[job(&queue, 100, false)]) + .await + .expect("per-row insert takes the lane"); + let batch = { + let pool = pool.clone(); + let queue = queue.clone(); + tokio::spawn(async move { + insert_many_skip_unique_conflicts(&pool, &[job(&queue, 0, true)]).await + }) + }; + tokio::time::sleep(Duration::from_millis(3_000)).await; + let per_row_outcome = tokio::time::timeout(WAIT_DEADLINE, async { + insert_many(&mut *per_row, &[job(&queue, 0, true)]).await?; + per_row.commit().await?; + Ok::<_, AwaError>(()) + }) + .await + .expect("the per-row producer's side of the cycle ended"); + let batch_outcome = tokio::time::timeout(WAIT_DEADLINE, batch) + .await + .expect("the batch's side of the cycle ended") + .expect("batch task"); + let aborted = [ + per_row_outcome + .as_ref() + .err() + .is_some_and(AwaError::is_deadlock), + batch_outcome + .as_ref() + .err() + .is_some_and(AwaError::is_deadlock), + ]; + assert_eq!( + aborted.iter().filter(|aborted| **aborted).count(), + 1, + "exactly one side is aborted: per-row {per_row_outcome:?}, batch {batch_outcome:?}" + ); +} diff --git a/awa/tests/insert_many_schema_fallback_test.rs b/awa/tests/insert_many_schema_fallback_test.rs index 275708b6d..44c0537f4 100644 --- a/awa/tests/insert_many_schema_fallback_test.rs +++ b/awa/tests/insert_many_schema_fallback_test.rs @@ -446,28 +446,33 @@ async fn copy_without_execute_on_the_batch_function_stays_per_row() { } } -/// A body from before the per-lane registration fix keeps the patch pending, -/// and `awa migrate` replaces it. +/// An earlier body keeps the patch pending, and `awa migrate` replaces it: +/// the first one, which registered every new lane before reserving any, and +/// the 0.6.15 one, which waited for a lane without bound while holding the +/// batch's unique claims. #[tokio::test] async fn an_earlier_batch_function_body_keeps_the_patch_pending() { - let db = ScratchDatabase::create().await; - let pool = &db.pool; - sqlx::raw_sql(include_str!( - "fixtures/insert_jobs_compat_batch_registers_lanes_first.sql" - )) - .execute(pool) - .await - .expect("install the earlier body"); - assert_eq!(pending_patches(pool).await, vec![PATCH_NAME]); - - let report = migrations::run_with_report(pool).await.expect("migrate"); - let patch = report - .patches - .iter() - .find(|patch| patch.name == PATCH_NAME) - .expect("patch report"); - assert_eq!(patch.outcome, migrations::PatchOutcome::Applied); - assert!(pending_patches(pool).await.is_empty()); - - db.drop().await; + for earlier_body in [ + include_str!("fixtures/insert_jobs_compat_batch_registers_lanes_first.sql"), + include_str!("fixtures/insert_jobs_compat_batch_unbounded_lane_wait.sql"), + ] { + let db = ScratchDatabase::create().await; + let pool = &db.pool; + sqlx::raw_sql(earlier_body) + .execute(pool) + .await + .expect("install the earlier body"); + assert_eq!(pending_patches(pool).await, vec![PATCH_NAME]); + + let report = migrations::run_with_report(pool).await.expect("migrate"); + let patch = report + .patches + .iter() + .find(|patch| patch.name == PATCH_NAME) + .expect("patch report"); + assert_eq!(patch.outcome, migrations::PatchOutcome::Applied); + assert!(pending_patches(pool).await.is_empty()); + + db.drop().await; + } } diff --git a/awa/tests/patch_upgrade_rehearsal_test.rs b/awa/tests/patch_upgrade_rehearsal_test.rs index d64df92a8..c5a63e8f6 100644 --- a/awa/tests/patch_upgrade_rehearsal_test.rs +++ b/awa/tests/patch_upgrade_rehearsal_test.rs @@ -754,8 +754,13 @@ async fn released_patch_upgrade_and_rollback() { .unwrap(); assert_eq!( pending.iter().map(|patch| patch.name).collect::>(), - ["claim_batch_job_index", "prune_controls"], - "a released 0.6.15 deployment needs the claim batch index fix and prune controls" + [ + "claim_batch_job_index", + "insert_jobs_compat_batch", + "prune_controls" + ], + "a released 0.6.15 deployment needs the claim batch index fix, the bounded-wait \ + batch function and prune controls" ); } @@ -824,13 +829,13 @@ 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(), 2); + assert_eq!(patch_sql.matches("-- Schema patch R__").count(), 3); assert!(!patch_sql.contains("R__prepared_lane_sequences")); 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!(patch_sql.contains("R__insert_jobs_compat_batch")); } assert!( !patch_sql.contains("-- Migration V"), @@ -975,6 +980,7 @@ async fn released_patch_upgrade_and_rollback() { } ), "Schema patch claim_batch_job_index: applied".to_string(), + "Schema patch insert_jobs_compat_batch: applied".to_string(), "Schema patch prune_controls: applied".to_string(), format!("Storage routing: unchanged (state active, engine queue_storage, schema {SCHEMA}"), ] { diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index d043b2732..7f3bb3e67 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -476,9 +476,9 @@ Cause: - on queue storage every enqueue holds a transaction-scoped advisory lock on its `(queue, priority, shard)` lane until the transaction commits; the `DETAIL` names two 64-bit advisory keys, and two transactions that took lanes in opposite orders hold one each and wait on the other's - `insert_many`, `insert_many_copy`, `insert_many_skip_unique_conflicts` and the `QueueStorage` batch enqueues register and lock their lanes one at a time in one canonical order, `(queue, priority, shard)` ascending with queue names compared bytewise, so two batch calls cannot hold each other's lanes in opposite orders - a 0.6.15 or earlier `QueueStorage` process (batch enqueue, or a worker promoting scheduled or retried jobs) creates a new `(queue, priority)` lane by taking its provisioning lock before writing its `queue_lanes` row; every other producer, and `QueueStorage` from 0.6.16, writes the row first, so one of each that are both the first to use the lane can deadlock. Upgrade those processes, or create a queue's lanes before traffic reaches it with `awa storage prepare-queue` -- with the `insert_jobs_compat_batch` schema patch, `insert_many_copy` and `insert_many_skip_unique_conflicts` take all of a batch's 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; a per-row transaction that holds a lane the batch needs and then inserts a job whose unique key the batch has claimed but not committed deadlocks with it, where it used to skip the job or fail with `23505` +- with the `insert_jobs_compat_batch` schema patch, `insert_many_copy` and `insert_many_skip_unique_conflicts` take all of a batch's 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. For 2 s the function waits at most 500 ms (or half of its session's `deadlock_timeout`, if lower) while it holds claims or lanes, then releases them and retries, so it gives way to a per-row transaction that holds a lane the batch needs and waits for one of its keys, provided that transaction's `deadlock_timeout` is above 500 ms (the default is 1 s) and the batch inserts its unique claims within 500 ms (10,000 to 20,000 keys on a fast server, fewer under load; split larger batches). Its last attempt waits as any statement does, so a per-row transaction that holds the lane for longer than the retries (about 2.5 s) before it inserts the key still deadlocks with it, and a cycle that retrying cannot break, for example through a lock the batch's own transaction took in an earlier statement, still ends in `40P01`, after 3 to 4 s and usually by aborting the batch's transaction. The 0.6.15 body of the function waited instead, and the two deadlocked; `awa migrate` replaces it (`awa migrate --sql --pending` prints `R__insert_jobs_compat_batch` while the old body is installed) - unique keys and the application's own locks can form other cycles -- a transaction that enqueues with several separate statements (`insert` per job, handler-staged follow-ups, `Transaction.insert` in a loop) takes locks in statement order and can deadlock against a batch or another such transaction that touches the same lanes in a different order +- a transaction that enqueues with several separate statements (`insert` per job, handler-staged follow-ups, `Transaction.insert` in a loop) takes locks in statement order and can deadlock against a generic `insert_many`, a `QueueStorage` batch enqueue or another such transaction that touches the same lanes in a different order; a batch through `awa.insert_jobs_compat_batch` normally gives way to it instead Fix: 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 aeba63010..b0a60c7b4 100644 --- a/docs/upgrade-0.6.15-to-0.6.16.md +++ b/docs/upgrade-0.6.15-to-0.6.16.md @@ -1,6 +1,6 @@ # Upgrading from 0.6.15 to 0.6.16 -0.6.16 adds a durable, database-wide halt for `maintenance.prune`. It is a rolling patch upgrade, but older binaries ignore the control and can still become maintenance leader and prune. Upgrade or fence every potential leader before relying on halt. +0.6.16 adds a durable, database-wide halt for `maintenance.prune`, and batch inserts no longer deadlock with per-row producers. It is a rolling patch upgrade, but older binaries ignore the control and can still become maintenance leader and prune. Upgrade or fence every potential leader before relying on halt. ## Upgrade @@ -12,7 +12,7 @@ awa --database-url "$AWA_MIGRATOR_DATABASE_URL" migrate ``` - The schema stays at v40. The output reports `Schema patch claim_batch_job_index: applied` and `Schema patch prune_controls: applied`. + The schema stays at v40. The output reports `Schema patch claim_batch_job_index: applied`, `Schema patch insert_jobs_compat_batch: applied` and `Schema patch prune_controls: applied`. The `insert_jobs_compat_batch` patch replaces the batch insert function's body; 0.6.15 producers get that fix as soon as the migration commits. 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. @@ -46,6 +46,27 @@ awa --database-url "$AWA_MIGRATOR_DATABASE_URL" migrate --extract-to ./sql/awa - Replace all `R__` files together and keep previously applied `V` files unchanged. `--pending` lists only patches the connected database lacks. +## Unique keys shared with per-row producers + +The schema stays at v40. `awa migrate` replaces the body of `awa.insert_jobs_compat_batch`, which `insert_many_copy` and `insert_many_skip_unique_conflicts` call. A 0.6.15 database reports the `insert_jobs_compat_batch` patch as pending until it does. The patch is still one `CREATE OR REPLACE FUNCTION` with no table DDL and no table locks, and the function's signature, privileges and results are unchanged. + +The function takes a batch's unique claims before its lane locks, while per-row inserts (generic `insert_many`, single-job inserts, and 0.6.14 and earlier binaries) take each job's claim and its lane together. In 0.6.15, a per-row transaction that held a lane the batch needed and then inserted a key the batch had claimed but not committed waited on the batch while the batch waited on it, and PostgreSQL aborted one of them with `deadlock detected` (SQLSTATE `40P01`). + +In 0.6.16 the function takes its claims and lanes in a subtransaction and, while it holds any of them, waits at most 500 ms per lock, or half of its own session's `deadlock_timeout` if that is lower. When a wait would run longer, the batch rolls back what it took and tries again after a short randomised pause. The per-row transaction gets the key or the lane, proceeds and commits, and on its next attempt the batch skips the keys it committed, or fails with a unique violation when it does not skip conflicts, as it would for any key committed before it. The same applies to a transaction that enqueues lanes in separate statements in the opposite order to a batch. Two batches through the function still take claims in key order and lanes in lane order, so they cannot deadlock with each other. + +After 2 s of retries, the batch makes one last attempt that waits as the 0.6.15 function did. Contention that outlasts the retries is then served in lock order, and a cycle that rolling back cannot break ends in `deadlock detected` rather than endless retries. The protection therefore lasts only for the bounded phase: a per-row transaction that holds a lane for longer than about 2.5 s and only then inserts one of the batch's keys meets the ordinary attempt, and one of the two is aborted with `40P01` as in 0.6.15. + +What callers see: + +- A batch insert and a per-row producer that share keys or lanes no longer abort each other with `40P01` when the batch call is the first thing its transaction locks and the per-row producer's `deadlock_timeout` is above 500 ms (PostgreSQL's default is 1 s). The batch cannot see another session's setting, so the cap is fixed; a per-row session configured at or below 500 ms can still be aborted by its own, earlier deadlock check. +- A per-row transaction that holds a lane longer than the bounded phase (about 2.5 s) before it inserts one of the batch's keys can still deadlock with the batch, as in 0.6.15. +- A batch whose transaction already holds locks from earlier statements can still be part of a deadlock: rolling back the batch cannot release them. That cycle now ends later than in 0.6.15, after about 3 to 4 s instead of 1 to 2 s, and the transaction usually aborted is the batch's caller rather than the per-row producer. +- A batch can only give way while inserting its unique claims takes less than the 500 ms budget. Each claim costs roughly 20 µs on a fast database server and more on a slow or busy one, so the limit is in the range of 10,000 to 20,000 unique keys. A larger batch, or one whose claims are slowed by load, uses up each bounded attempt before it reaches its lanes and ends in the ordinary attempt, so it behaves as in 0.6.15. Split such batches if they share keys or lanes with per-row producers. +- Under contention that outlasts the retries, a batch waits for its lanes in lock order, as in 0.6.15. +- A caller's `lock_timeout` applies to each of the batch's lock waits, as it does to any statement's, and a wait it ends fails the call with `lock_not_available` (SQLSTATE `55P03`). Time the batch spends on its own work or on retries does not count against it. +- The 500 ms budget applies to each wait, not to a statement: a claim statement that meets several keys still in flight, or a lane reservation that waits for the lane's provisioning lock and then its lock, can wait up to the remaining budget each time. +- The lane lock is still held only for the lane's reservation, the ready-row writes and the commit. + ## Halt, inspect and resume Only `maintenance.prune` with mode `drain` is supported. It stops compatible runtimes from admitting reclamation and housekeeping, including descriptor publication; work admitted before observation may finish. Enqueue, dispatch, handlers, cron, promotion, rescue, admin batches and queue health gauges continue. Fresh storage initialization, auto-finalization and storage selection proceed even during halt or an initial control-read failure. @@ -74,3 +95,5 @@ 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 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. + +For the batch insert fix, roll binaries back to 0.6.15 at any time and leave the patch in place: 0.6.15 binaries call the same function with the same arguments. Rolling the function back means applying the 0.6.15 `R__insert_jobs_compat_batch.sql`, which brings the deadlock back.