From 0cb61a66798e58492dd60b626af26bd13b09282a Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Fri, 6 Nov 2026 13:27:37 +1300 Subject: [PATCH 1/5] fix(queue-storage): create queue lanes as the schema owner (v053) A lane's enqueue and claim sequences were created by whichever role first touched the lane, through SECURITY INVOKER helpers, so only that role and the roles its default privileges named could use them. Another runtime role then failed with "permission denied for sequence" on enqueue, claim or promotion. Each queue-storage schema now has two SECURITY DEFINER functions owned by the schema owner, provision_lane and provision_lane_sequences. The lane helpers create a missing sequence through them, and each new sequence gets the privileges of the head table it backs. ensure_lane_inserts calls provision_lane when it exists and is executable, inside a savepoint, and otherwise writes the lane rows as before. v053 refreshes the helpers and repairs existing lane sequences one at a time, reassigning and granting only what the migrating role may and reporting the rest. awa.lane_sequence_problems and the new `awa storage lane-sequences` command list what remains with the statements to run; awa.repair_lane_sequences and `--repair` rerun the repair. --- CHANGELOG.md | 24 + awa-cli/src/main.rs | 59 +- .../v023_install_queue_storage_substrate.sql | 416 +++++++++- .../v053_definer_lane_provisioning.sql | 63 ++ awa-model/src/migrations.rs | 8 +- awa-model/src/queue_storage.rs | 106 ++- awa-model/src/storage.rs | 76 ++ .../tests/least_privilege_runtime_test.rs | 41 +- .../fixtures/v052_lane_sequence_helpers.sql | 190 +++++ awa/tests/lane_ownership_roles_test.rs | 735 ++++++++++++++++++ docs/queue-storage-substrate.md | 2 +- docs/security/database-roles.md | 50 +- docs/upgrade-0.6-to-0.7.md | 39 +- skills/awa-operations/SKILL.md | 12 +- 14 files changed, 1734 insertions(+), 87 deletions(-) create mode 100644 awa-model/migrations/v053_definer_lane_provisioning.sql create mode 100644 awa/tests/fixtures/v052_lane_sequence_helpers.sql create mode 100644 awa/tests/lane_ownership_roles_test.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 065d50db1..6780d6113 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,30 @@ Notable changes between releases. Detailed migration notes for storage transitio ### Fixed +- **A queue lane is usable by every runtime role, whichever role creates it + first (migration v053).** A lane's enqueue and claim sequences belonged to + the role that first enqueued into, claimed from or promoted into it, so with + several runtime roles sharing a group role, any role other than the creator + whose default privileges did not cover it failed with `permission denied for + sequence`. That included the maintenance leader promoting into a shard + created after `enqueue_shards` was raised. Lanes are now created through two + `SECURITY DEFINER` functions per queue-storage schema, `provision_lane` and + `provision_lane_sequences`, owned by the schema owner. Each new sequence gets + the privileges of the head table it backs, explicitly rather than through + default privileges. A runtime without schema `CREATE` can therefore create + lanes itself, and `awa storage prepare-queue` is optional and may run as any + runtime role. v053 gives existing lane sequences to the schema owner and adds + their missing grants, one sequence at a time and only where the migrating + role is allowed to. It never fails on a sequence it cannot reassign. The new + read-only `awa storage lane-sequences` (or + `awa.lane_sequence_problems(schema)`) lists what remains, with the statements + to run and the roles to run them as, and exits 2 while any remain; + `--repair` (or `awa.repair_lane_sequences(schema)`) repairs what the + connected role may. `awa migrate` prints a summary of the leftovers. A binary + running ahead of v053 detects that `provision_lane` is missing and creates + lanes as before, and older binaries create correctly owned lanes once v053 + has refreshed the helpers they call. See "Lane ownership and v053" in the + 0.6 to 0.7 upgrade guide. - **Batch enqueues take their lane locks in one order, so they no longer deadlock with each other.** On queue storage every enqueue holds a transaction-scoped advisory lock on its `(queue, priority, shard)` lane diff --git a/awa-cli/src/main.rs b/awa-cli/src/main.rs index 405fc3023..5e53264c5 100644 --- a/awa-cli/src/main.rs +++ b/awa-cli/src/main.rs @@ -328,11 +328,13 @@ fn access_for(command: &Commands) -> context::Access { }, Commands::Storage { command } => match command { StorageCommands::Status => ReadOnly, + StorageCommands::LaneSequences { repair: false, .. } => ReadOnly, StorageCommands::Finalize { check: true, .. } => ReadOnly, StorageCommands::FlipRingAuthority { check: true, .. } => ReadOnly, StorageCommands::Prepare { .. } | StorageCommands::PrepareQueueStorageSchema { .. } | StorageCommands::PrepareQueue { .. } + | StorageCommands::LaneSequences { .. } | StorageCommands::Abort | StorageCommands::EnterMixedTransition { .. } | StorageCommands::Finalize { .. } @@ -811,7 +813,7 @@ enum StorageCommands { #[arg(long)] reset: bool, }, - /// Provision lane sequences as the migrator, before starting a runtime without DDL + /// Provision every priority and shard of a queue ahead of its first enqueue PrepareQueue { /// Queue-storage schema to provision #[arg(long, default_value = "awa")] @@ -826,6 +828,16 @@ enum StorageCommands { #[arg(long, default_value_t = 1)] queue_stripe_count: usize, }, + /// List lane sequences not owned by the schema owner or missing runtime + /// grants, with the statements that fix them. Exits 2 while any remain. + LaneSequences { + /// Queue-storage schema to check + #[arg(long, default_value = "awa")] + schema: String, + /// Reassign and grant what the connected role is allowed to first + #[arg(long)] + repair: bool, + }, /// Abort a prepared or mixed-transition storage rollout before final activation Abort, /// Enter mixed transition and begin routing new writes to the prepared engine @@ -1182,6 +1194,16 @@ async fn run() -> Result<(), Box> { }; awa_model::migrations::run_with_options(&pool, options).await?; println!("Migrations applied successfully."); + for (schema, count) in + awa_model::storage::lane_sequence_problem_counts(&pool).await? + { + eprintln!( + "warning: {count} lane sequence(s) in schema {schema} are not owned by \ + the schema owner or lack runtime grants, and the migrating role could \ + not repair them. Run `awa storage lane-sequences --schema {schema}` for \ + the statements to run and the roles to run them as." + ); + } } } @@ -1933,6 +1955,41 @@ async fn run() -> Result<(), Box> { }))? ); } + StorageCommands::LaneSequences { schema, repair } => { + let remaining = if repair { + let results = + awa_model::storage::repair_lane_sequences(&pool, &schema).await?; + println!( + "{}", + serde_json::to_string_pretty(&serde_json::json!({ + "schema": schema, "results": results, + }))? + ); + results + .iter() + .filter(|result| result.outcome != "repaired") + .count() + } else { + let problems = + awa_model::storage::lane_sequence_problems(&pool, &schema).await?; + println!( + "{}", + serde_json::to_string_pretty(&serde_json::json!({ + "schema": schema, "problems": problems, + }))? + ); + problems.len() + }; + if remaining > 0 { + eprintln!( + "{remaining} lane sequence(s) in schema {schema} need repair. Run \ + each missing_grants statement as the sequence owner (or a role \ + that can SET ROLE to it), and each reassign_sql statement as a \ + role that can SET ROLE to both the owner and expected_owner." + ); + std::process::exit(2); + } + } StorageCommands::Abort => { awa_model::storage::abort(&pool).await?; let report = awa_model::storage::status_report(&pool).await?; diff --git a/awa-model/migrations/v023_install_queue_storage_substrate.sql b/awa-model/migrations/v023_install_queue_storage_substrate.sql index d852b2ac5..b5526cccc 100644 --- a/awa-model/migrations/v023_install_queue_storage_substrate.sql +++ b/awa-model/migrations/v023_install_queue_storage_substrate.sql @@ -42,6 +42,12 @@ LANGUAGE plpgsql SECURITY INVOKER SET search_path = pg_catalog AS $refresh$ +DECLARE + v_schema_owner OID; + v_function REGPROCEDURE; + v_function_owner OID; + v_set_mode TEXT := CASE WHEN current_setting('server_version_num')::int >= 160000 + THEN 'SET' ELSE 'MEMBER' END; BEGIN EXECUTE format( $ddl$ @@ -69,28 +75,40 @@ BEGIN ); BEGIN IF to_regclass(format('%%I.%%I', %1$L, v_enqueue_seq)) IS NULL THEN - IF NOT has_schema_privilege(current_user, %1$L, 'CREATE') THEN + IF has_function_privilege( + to_regprocedure(format('%%I.provision_lane_sequences(text,smallint,smallint)', %1$L)), + 'EXECUTE' + ) THEN + PERFORM %1$I.provision_lane_sequences(p_queue, p_priority, p_enqueue_shard); + ELSIF NOT has_schema_privilege(current_user, %1$L, 'CREATE') THEN RAISE EXCEPTION 'lane sequence %%.%% has not been provisioned', %1$L, v_enqueue_seq USING ERRCODE = '42501', - HINT = 'Run awa storage prepare-queue as the migrator before starting producers or workers.'; + HINT = 'Run awa migrate, or awa storage prepare-queue as the migrator, before starting producers or workers.'; + ELSE + EXECUTE format( + 'CREATE SEQUENCE IF NOT EXISTS %%I.%%I AS bigint START WITH 1 MINVALUE 1 CACHE 1', + %1$L, + v_enqueue_seq + ); END IF; - EXECUTE format( - 'CREATE SEQUENCE IF NOT EXISTS %%I.%%I AS bigint START WITH 1 MINVALUE 1 CACHE 1', - %1$L, - v_enqueue_seq - ); END IF; IF to_regclass(format('%%I.%%I', %1$L, v_claim_seq)) IS NULL THEN - IF NOT has_schema_privilege(current_user, %1$L, 'CREATE') THEN + IF has_function_privilege( + to_regprocedure(format('%%I.provision_lane_sequences(text,smallint,smallint)', %1$L)), + 'EXECUTE' + ) THEN + PERFORM %1$I.provision_lane_sequences(p_queue, p_priority, p_enqueue_shard); + ELSIF NOT has_schema_privilege(current_user, %1$L, 'CREATE') THEN RAISE EXCEPTION 'lane sequence %%.%% has not been provisioned', %1$L, v_claim_seq USING ERRCODE = '42501', - HINT = 'Run awa storage prepare-queue as the migrator before starting producers or workers.'; + HINT = 'Run awa migrate, or awa storage prepare-queue as the migrator, before starting producers or workers.'; + ELSE + EXECUTE format( + 'CREATE SEQUENCE IF NOT EXISTS %%I.%%I AS bigint START WITH 1 MINVALUE 1 CACHE 1', + %1$L, + v_claim_seq + ); END IF; - EXECUTE format( - 'CREATE SEQUENCE IF NOT EXISTS %%I.%%I AS bigint START WITH 1 MINVALUE 1 CACHE 1', - %1$L, - v_claim_seq - ); END IF; UPDATE %1$I.queue_enqueue_heads @@ -131,16 +149,22 @@ BEGIN v_start BIGINT; BEGIN IF to_regclass(format('%%I.%%I', %1$L, v_seq_name)) IS NULL THEN - IF NOT has_schema_privilege(current_user, %1$L, 'CREATE') THEN + IF has_function_privilege( + to_regprocedure(format('%%I.provision_lane_sequences(text,smallint,smallint)', %1$L)), + 'EXECUTE' + ) THEN + PERFORM %1$I.provision_lane_sequences(NEW.queue, NEW.priority, NEW.enqueue_shard); + ELSIF NOT has_schema_privilege(current_user, %1$L, 'CREATE') THEN RAISE EXCEPTION 'lane sequence %%.%% has not been provisioned', %1$L, v_seq_name USING ERRCODE = '42501', - HINT = 'Run awa storage prepare-queue as the migrator before starting producers or workers.'; + HINT = 'Run awa migrate, or awa storage prepare-queue as the migrator, before starting producers or workers.'; + ELSE + EXECUTE format( + 'CREATE SEQUENCE IF NOT EXISTS %%I.%%I AS bigint START WITH 1 MINVALUE 1 CACHE 1', + %1$L, + v_seq_name + ); END IF; - EXECUTE format( - 'CREATE SEQUENCE IF NOT EXISTS %%I.%%I AS bigint START WITH 1 MINVALUE 1 CACHE 1', - %1$L, - v_seq_name - ); END IF; NEW.seq_name := v_seq_name; @@ -197,16 +221,22 @@ BEGIN ); BEGIN IF to_regclass(format('%%I.%%I', %1$L, v_seq_name)) IS NULL THEN - IF NOT has_schema_privilege(current_user, %1$L, 'CREATE') THEN + IF has_function_privilege( + to_regprocedure(format('%%I.provision_lane_sequences(text,smallint,smallint)', %1$L)), + 'EXECUTE' + ) THEN + PERFORM %1$I.provision_lane_sequences(NEW.queue, NEW.priority, NEW.enqueue_shard); + ELSIF NOT has_schema_privilege(current_user, %1$L, 'CREATE') THEN RAISE EXCEPTION 'lane sequence %%.%% has not been provisioned', %1$L, v_seq_name USING ERRCODE = '42501', - HINT = 'Run awa storage prepare-queue as the migrator before starting producers or workers.'; + HINT = 'Run awa migrate, or awa storage prepare-queue as the migrator, before starting producers or workers.'; + ELSE + EXECUTE format( + 'CREATE SEQUENCE IF NOT EXISTS %%I.%%I AS bigint START WITH 1 MINVALUE 1 CACHE 1', + %1$L, + v_seq_name + ); END IF; - EXECUTE format( - 'CREATE SEQUENCE IF NOT EXISTS %%I.%%I AS bigint START WITH 1 MINVALUE 1 CACHE 1', - %1$L, - v_seq_name - ); END IF; NEW.seq_name := v_seq_name; @@ -221,11 +251,341 @@ BEGIN p_schema ); + -- Lane sequences are created by the schema owner through these two + -- SECURITY DEFINER functions, whichever role first touches the lane, so + -- every runtime role can use them. Their grants mirror the head table each + -- sequence backs (see awa.lane_sequence_problems); they never depend on + -- the creating role's default privileges. + EXECUTE format( + $ddl$ + CREATE OR REPLACE FUNCTION %1$I.provision_lane_sequences( + p_queue TEXT, + p_priority SMALLINT, + p_enqueue_shard SMALLINT + ) + RETURNS VOID + LANGUAGE plpgsql + SECURITY DEFINER + SET search_path = pg_catalog, pg_temp + AS $func$ + DECLARE + v_seq_name TEXT; + v_grant TEXT; + BEGIN + IF p_queue IS NULL OR p_queue = '' + OR p_priority IS NULL OR p_priority NOT BETWEEN 1 AND 4 + OR p_enqueue_shard IS NULL OR p_enqueue_shard NOT BETWEEN 0 AND 63 THEN + RAISE EXCEPTION 'invalid queue lane: queue %%, priority %%, enqueue shard %%', + quote_nullable(p_queue), p_priority, p_enqueue_shard + USING ERRCODE = '22023'; + END IF; + + -- The lane lock QueueStorage takes before it writes lane rows, so + -- concurrent first touches of one lane create its sequences once. + PERFORM pg_advisory_xact_lock(hashtextextended( + format('%%s:%%s:%%s:%%s', %1$L, p_queue, p_priority, p_enqueue_shard), + 0 + )); + + FOREACH v_seq_name IN ARRAY ARRAY[ + %1$I.queue_lane_sequence_name('queue_enqueue_seq', p_queue, p_priority, p_enqueue_shard), + %1$I.queue_lane_sequence_name('queue_claim_seq', p_queue, p_priority, p_enqueue_shard) + ] LOOP + IF to_regclass(format('%%I.%%I', %1$L, v_seq_name)) IS NULL THEN + EXECUTE format( + 'CREATE SEQUENCE IF NOT EXISTS %%I.%%I AS bigint START WITH 1 MINVALUE 1 CACHE 1', + %1$L, + v_seq_name + ); + FOR v_grant IN + SELECT unnest(problem.missing_grants) + FROM awa.lane_sequence_problems(%1$L, v_seq_name) AS problem + WHERE problem.owner = current_user + LOOP + EXECUTE v_grant; + END LOOP; + END IF; + END LOOP; + END; + $func$ + $ddl$, + p_schema + ); + + EXECUTE format( + $ddl$ + CREATE OR REPLACE FUNCTION %1$I.provision_lane( + p_queue TEXT, + p_priority SMALLINT, + p_enqueue_shard SMALLINT + ) + RETURNS VOID + LANGUAGE plpgsql + SECURITY DEFINER + SET search_path = pg_catalog, pg_temp + AS $func$ + BEGIN + -- Validates the lane and takes its lock before any row is written. + PERFORM %1$I.provision_lane_sequences(p_queue, p_priority, p_enqueue_shard); + + INSERT INTO %1$I.queue_lanes (queue, priority) + VALUES (p_queue, p_priority) + ON CONFLICT (queue, priority) DO NOTHING; + + -- The head triggers fire before ON CONFLICT and set the lane's + -- sequence, so rows that already exist are skipped outright. + INSERT INTO %1$I.queue_enqueue_heads (queue, priority, enqueue_shard) + SELECT p_queue, p_priority, p_enqueue_shard + WHERE NOT EXISTS ( + SELECT 1 FROM %1$I.queue_enqueue_heads AS heads + WHERE heads.queue = p_queue + AND heads.priority = p_priority + AND heads.enqueue_shard = p_enqueue_shard + ) + ON CONFLICT (queue, priority, enqueue_shard) DO NOTHING; + + INSERT INTO %1$I.queue_claim_heads (queue, priority, enqueue_shard) + SELECT p_queue, p_priority, p_enqueue_shard + WHERE NOT EXISTS ( + SELECT 1 FROM %1$I.queue_claim_heads AS heads + WHERE heads.queue = p_queue + AND heads.priority = p_priority + AND heads.enqueue_shard = p_enqueue_shard + ) + ON CONFLICT (queue, priority, enqueue_shard) DO NOTHING; + + PERFORM %1$I.ensure_lane_sequences(p_queue, p_priority, p_enqueue_shard); + END; + $func$ + $ddl$, + p_schema + ); + + SELECT namespace.nspowner INTO v_schema_owner + FROM pg_namespace AS namespace + WHERE namespace.nspname = p_schema; + + FOREACH v_function IN ARRAY ARRAY[ + format('%I.provision_lane_sequences(text,smallint,smallint)', p_schema)::regprocedure, + format('%I.provision_lane(text,smallint,smallint)', p_schema)::regprocedure + ] LOOP + EXECUTE format('GRANT EXECUTE ON FUNCTION %s TO PUBLIC', v_function); + + SELECT routine.proowner INTO v_function_owner + FROM pg_proc AS routine + WHERE routine.oid = v_function; + + IF v_function_owner <> v_schema_owner THEN + IF pg_has_role(current_user, v_function_owner, 'USAGE') + AND pg_has_role(current_user, v_schema_owner, v_set_mode) THEN + EXECUTE format( + 'ALTER FUNCTION %s OWNER TO %I', + v_function, + pg_get_userbyid(v_schema_owner) + ); + ELSE + RAISE WARNING '% is owned by %, not the owner % of schema %; lane sequences it creates will be owned by % too', + v_function, pg_get_userbyid(v_function_owner), pg_get_userbyid(v_schema_owner), + p_schema, pg_get_userbyid(v_function_owner) + USING HINT = format( + 'Run awa migrate as a role that can SET ROLE to %s, or as %s itself.', + quote_ident(pg_get_userbyid(v_schema_owner)), + quote_ident(pg_get_userbyid(v_schema_owner)) + ); + END IF; + END IF; + END LOOP; + END; $refresh$; REVOKE EXECUTE ON FUNCTION awa.refresh_lane_sequence_helpers(TEXT) FROM PUBLIC; +-- Lane sequences in p_schema (or the one named p_sequence) that are not owned +-- by the schema owner or lack a runtime grant. A lane sequence carries a head +-- table's cursor, so whoever may write that head table (INSERT or UPDATE) gets +-- USAGE, SELECT and UPDATE on it, and whoever may read it gets SELECT: +-- queue_enqueue_seq_* mirror queue_enqueue_heads, queue_claim_seq_* mirror +-- queue_claim_heads. missing_grants holds the statements to run as the +-- sequence owner; reassign_sql, when the owner is wrong, needs a role that can +-- SET ROLE to both the current owner and the schema owner. Read-only. +CREATE OR REPLACE FUNCTION awa.lane_sequence_problems(p_schema TEXT, p_sequence TEXT DEFAULT NULL) +RETURNS TABLE ( + sequence_name TEXT, + owner TEXT, + expected_owner TEXT, + missing_grants TEXT[], + reassign_sql TEXT +) +LANGUAGE sql +STABLE +SECURITY INVOKER +SET search_path = pg_catalog, pg_temp +AS $problems$ + WITH target AS ( + SELECT namespace.oid, namespace.nspowner + FROM pg_namespace AS namespace + WHERE namespace.nspname = p_schema + ), + lane_sequence AS ( + SELECT sequence.oid, + sequence.relname::text AS relname, + sequence.relowner, + sequence.relacl, + target.nspowner, + CASE WHEN starts_with(sequence.relname::text, 'queue_enqueue_seq_') + THEN 'queue_enqueue_heads' ELSE 'queue_claim_heads' END AS head_table + FROM target + JOIN pg_class AS sequence ON sequence.relnamespace = target.oid + WHERE sequence.relkind = 'S' + AND sequence.relname ~ '^queue_(enqueue|claim)_seq_[0-9a-f]{32}$' + AND (p_sequence IS NULL OR sequence.relname = p_sequence) + ), + wanted AS ( + SELECT DISTINCT lane_sequence.oid, head_privilege.grantee, sequence_privilege.privilege_type + FROM lane_sequence + JOIN pg_class AS head + ON head.relnamespace = (SELECT target.oid FROM target) + AND head.relname = lane_sequence.head_table + CROSS JOIN LATERAL aclexplode(COALESCE(head.relacl, acldefault('r', head.relowner))) AS head_privilege + CROSS JOIN LATERAL unnest( + CASE head_privilege.privilege_type + WHEN 'SELECT' THEN ARRAY['SELECT'] + WHEN 'INSERT' THEN ARRAY['USAGE', 'SELECT', 'UPDATE'] + WHEN 'UPDATE' THEN ARRAY['USAGE', 'SELECT', 'UPDATE'] + ELSE ARRAY[]::TEXT[] + END + ) AS sequence_privilege(privilege_type) + WHERE head_privilege.grantee <> lane_sequence.nspowner + ), + missing AS ( + SELECT wanted.oid, wanted.grantee, wanted.privilege_type + FROM wanted + JOIN lane_sequence ON lane_sequence.oid = wanted.oid + WHERE NOT EXISTS ( + SELECT 1 + FROM aclexplode(COALESCE(lane_sequence.relacl, acldefault('S', lane_sequence.relowner))) AS held + WHERE held.grantee = wanted.grantee + AND held.privilege_type = wanted.privilege_type + ) + ) + SELECT lane_sequence.relname, + pg_get_userbyid(lane_sequence.relowner)::text, + pg_get_userbyid(lane_sequence.nspowner)::text, + ARRAY( + SELECT format( + 'GRANT %s ON SEQUENCE %I.%I TO %s', + string_agg(missing.privilege_type, ', ' ORDER BY missing.privilege_type DESC), + p_schema, + lane_sequence.relname, + CASE missing.grantee WHEN 0 THEN 'PUBLIC' + ELSE quote_ident(pg_get_userbyid(missing.grantee)) END + ) + FROM missing + WHERE missing.oid = lane_sequence.oid + GROUP BY missing.grantee + ORDER BY 1 + ), + CASE WHEN lane_sequence.relowner <> lane_sequence.nspowner THEN format( + 'ALTER SEQUENCE %I.%I OWNER TO %I', + p_schema, lane_sequence.relname, pg_get_userbyid(lane_sequence.nspowner) + ) END + FROM lane_sequence + WHERE lane_sequence.relowner <> lane_sequence.nspowner + OR EXISTS (SELECT 1 FROM missing WHERE missing.oid = lane_sequence.oid) + ORDER BY lane_sequence.relname +$problems$; + +-- Give every lane sequence in p_schema to the schema owner and add its missing +-- runtime grants, one sequence at a time, doing only what the calling role is +-- allowed to: a sequence it cannot reassign keeps its owner and still gets +-- the grants when the caller can act as that owner. Never raises for a +-- sequence it cannot repair. Returns one row per sequence that had a problem: +-- outcome is 'repaired', 'owner_unchanged' (grants complete, owner still +-- wrong) or 'unrepaired', with the statements still to run. +CREATE OR REPLACE FUNCTION awa.repair_lane_sequences(p_schema TEXT) +RETURNS TABLE ( + sequence_name TEXT, + owner TEXT, + expected_owner TEXT, + outcome TEXT, + missing_grants TEXT[], + reassign_sql TEXT +) +LANGUAGE plpgsql +SECURITY INVOKER +SET search_path = pg_catalog, pg_temp +AS $repair$ +DECLARE + v_problem RECORD; + v_remaining RECORD; + v_has_remaining BOOLEAN; + v_owner OID; + v_set_mode TEXT := CASE WHEN current_setting('server_version_num')::int >= 160000 + THEN 'SET' ELSE 'MEMBER' END; + v_statement TEXT; +BEGIN + FOR v_problem IN + SELECT * FROM awa.lane_sequence_problems(p_schema) AS problem + LOOP + SELECT sequence.relowner INTO v_owner + FROM pg_class AS sequence + WHERE sequence.oid = to_regclass(format('%I.%I', p_schema, v_problem.sequence_name)); + + -- ALTER ... OWNER TO needs the current owner's privileges, SET ROLE to + -- the new owner, and the new owner's CREATE on the schema (which a + -- schema owner has). Check first so an unrelated owner is skipped + -- rather than raising. + IF v_problem.reassign_sql IS NOT NULL + AND pg_has_role(current_user, v_owner, 'USAGE') + AND pg_has_role(current_user, to_regrole(quote_ident(v_problem.expected_owner)), v_set_mode) THEN + BEGIN + EXECUTE v_problem.reassign_sql; + EXCEPTION WHEN insufficient_privilege THEN + NULL; + END; + END IF; + + SELECT * INTO v_remaining + FROM awa.lane_sequence_problems(p_schema, v_problem.sequence_name) AS problem; + v_has_remaining := FOUND; + + IF v_has_remaining + AND cardinality(v_remaining.missing_grants) > 0 + AND pg_has_role(current_user, to_regrole(quote_ident(v_remaining.owner)), 'USAGE') THEN + FOREACH v_statement IN ARRAY v_remaining.missing_grants LOOP + BEGIN + EXECUTE v_statement; + EXCEPTION WHEN insufficient_privilege THEN + NULL; + END; + END LOOP; + SELECT * INTO v_remaining + FROM awa.lane_sequence_problems(p_schema, v_problem.sequence_name) AS problem; + v_has_remaining := FOUND; + END IF; + + sequence_name := v_problem.sequence_name; + expected_owner := v_problem.expected_owner; + IF v_has_remaining THEN + owner := v_remaining.owner; + missing_grants := v_remaining.missing_grants; + reassign_sql := v_remaining.reassign_sql; + outcome := CASE WHEN cardinality(v_remaining.missing_grants) = 0 + THEN 'owner_unchanged' ELSE 'unrepaired' END; + ELSE + owner := v_problem.expected_owner; + missing_grants := ARRAY[]::TEXT[]; + reassign_sql := NULL; + outcome := 'repaired'; + END IF; + RETURN NEXT; + END LOOP; +END +$repair$; + +REVOKE EXECUTE ON FUNCTION awa.repair_lane_sequences(TEXT) FROM PUBLIC; + -- Retained job output (ADR-048): add the output columns to an existing -- substrate and expose them through terminal_jobs. Called by the installer -- and by migration v052, which does not replay the installer. Each column is diff --git a/awa-model/migrations/v053_definer_lane_provisioning.sql b/awa-model/migrations/v053_definer_lane_provisioning.sql new file mode 100644 index 000000000..a56d96978 --- /dev/null +++ b/awa-model/migrations/v053_definer_lane_provisioning.sql @@ -0,0 +1,63 @@ +-- v053: Create queue lanes as the schema owner, whichever role touches them first. +-- Lane sequences used to belong to the role that first enqueued into, claimed +-- from or promoted into a lane, and only that role (plus whoever its default +-- privileges named) could use them. refresh_lane_sequence_helpers now installs +-- two SECURITY DEFINER functions, provision_lane and provision_lane_sequences, +-- owned by the schema owner, and the lane helpers create a missing sequence +-- through them. Each new sequence gets the grants of the head table it backs. +-- This migration then repairs existing lane sequences: it gives each one to +-- the schema owner and adds its missing grants, one sequence at a time and +-- only where the migrating role may. Sequences it cannot reassign are reported +-- in a warning and by awa.lane_sequence_problems(); the migration still commits. +-- +-- N-1 binaries call the same helpers with the same arguments and cursor +-- semantics, so lanes they create are owned by the schema owner too. A binary +-- that predates provision_lane keeps using the helpers; this binary probes for +-- provision_lane and uses the helpers when it is missing. +-- +-- Live load: function replacement, plus ALTER SEQUENCE OWNER on the lane +-- sequences that need it. That takes each such sequence's ACCESS EXCLUSIVE +-- lock until commit, so enqueues and claims on those lanes wait for the +-- migration to commit; correctly owned lanes are not touched. Reads the +-- catalog only; no table DDL or row scans. + +DO $$ +DECLARE + v_schema TEXT; + v_leftovers TEXT; + v_count BIGINT; +BEGIN + FOR v_schema IN + SELECT discovered.schema_name FROM awa.queue_storage_schemas() AS discovered + LOOP + IF to_regclass(format('%I.queue_enqueue_heads', v_schema)) IS NULL + OR to_regclass(format('%I.queue_claim_heads', v_schema)) IS NULL + OR to_regclass(format('%I.queue_ring_state', v_schema)) IS NULL + OR to_regclass(format('%I.lease_ring_state', v_schema)) IS NULL + OR to_regclass(format('%I.claim_ring_state', v_schema)) IS NULL THEN + CONTINUE; + END IF; + + PERFORM awa.refresh_lane_sequence_helpers(v_schema); + + SELECT count(*), + string_agg(format('%s (owner %s, %s)', repair.sequence_name, repair.owner, repair.outcome), + ', ' ORDER BY repair.sequence_name) + INTO v_count, v_leftovers + FROM awa.repair_lane_sequences(v_schema) AS repair + WHERE repair.outcome <> 'repaired'; + + IF v_count > 0 THEN + RAISE WARNING '% lane sequence(s) in schema % still need repair: %', v_count, v_schema, v_leftovers + USING HINT = format( + 'Run awa storage lane-sequences --schema %s for the statements to run and the role to run them as.', + v_schema + ); + END IF; + END LOOP; +END; +$$; + +INSERT INTO awa.schema_version (version, description) +VALUES (53, 'Create queue lanes as the schema owner') +ON CONFLICT (version) DO NOTHING; diff --git a/awa-model/src/migrations.rs b/awa-model/src/migrations.rs index 385035b65..51c423222 100644 --- a/awa-model/src/migrations.rs +++ b/awa-model/src/migrations.rs @@ -5,7 +5,7 @@ use sqlx::{Connection, PgPool}; use tracing::{info, warn}; /// Current schema version. -pub const CURRENT_VERSION: i32 = 52; +pub const CURRENT_VERSION: i32 = 53; /// Migrations that require an exclusive (no-live-runtime) upgrade window. /// @@ -292,6 +292,11 @@ const MIGRATIONS: &[(i32, &str, &[&str])] = &[ ), (51, "Per-slot claim functions", &[V23_UP, V51_UP]), (52, "Retained job output", &[V23_UP, V52_UP]), + ( + 53, + "Create queue lanes as the schema owner", + &[V23_UP, V53_UP], + ), ]; const V1_UP: &str = include_str!("../migrations/v001_canonical_schema.sql"); @@ -352,6 +357,7 @@ const V49_UP: &str = include_str!("../migrations/v049_prepared_lane_sequences.sq const V50_UP: &str = include_str!("../migrations/v050_drop_ready_lane_shard_index.sql"); const V51_UP: &str = include_str!("../migrations/v051_claim_slot_functions.sql"); const V52_UP: &str = include_str!("../migrations/v052_retained_job_output.sql"); +const V53_UP: &str = include_str!("../migrations/v053_definer_lane_provisioning.sql"); /// Old version numbers from pre-0.4 releases that used V3/V4/V5 numbering. /// Also tolerates the unreleased inline-V6 branch numbering used during review. diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index 339001705..dafbbaa14 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -13,7 +13,7 @@ use sqlx::{PgPool, Postgres}; use std::collections::hash_map::DefaultHasher; use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use std::hash::{Hash, Hasher}; -use std::sync::atomic::{AtomicU16, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU16, AtomicUsize, Ordering}; use std::sync::Mutex; use std::time::{Duration, Instant}; use uuid::Uuid; @@ -3461,6 +3461,12 @@ pub struct QueueStorage { /// TRUNCATEs queue_lanes; `advance_enqueue_head` repairs a stale entry /// (head row gone after a rolled-back ensure_lane). ensured_lanes: Mutex>, + /// Whether `{schema}.provision_lane` (migration v053) was callable the + /// last time a lane was created. Only a positive answer is kept: lane + /// creation is rare, and the probe rides on the lane-lock round trip. + lane_provisioner_present: AtomicBool, + /// Set once a provisioner failure has been logged. + lane_provisioner_warned: AtomicBool, slot_fill_observations: Mutex>, prune_lock_timeout: Duration, cancel_phase_hook: CancelPhaseHookSlot, @@ -3516,6 +3522,8 @@ impl QueueStorage { enqueue_shards_cache: Mutex::new(HashMap::new()), claimer_gates: Mutex::new(HashMap::new()), ensured_lanes: Mutex::new(HashSet::new()), + lane_provisioner_present: AtomicBool::new(false), + lane_provisioner_warned: AtomicBool::new(false), slot_fill_observations: Mutex::new(HashMap::new()), prune_lock_timeout: DEFAULT_PRUNE_LOCK_TIMEOUT, cancel_phase_hook: CancelPhaseHookSlot(None), @@ -4594,9 +4602,10 @@ impl QueueStorage { .await } - /// Provision every priority and shard of a queue before deploying a runtime - /// role without schema CREATE. Run as the schema owner/migrator. This does - /// not change enqueue routing or the configured shard count. + /// Provision every priority and shard of a queue ahead of its first + /// enqueue. From migration v053 any runtime role may run it: the lanes are + /// created as the schema owner. This does not change enqueue routing or + /// the configured shard count. pub async fn prepare_queue( &self, pool: &PgPool, @@ -4637,11 +4646,35 @@ impl QueueStorage { ) -> Result<(), AwaError> { let schema = self.schema(); let lane_lock_key = format!("{schema}:{queue}:{priority}:{enqueue_shard}"); - sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))") - .bind(lane_lock_key) - .execute(tx.as_mut()) + let provisioner_callable = if self.lane_provisioner_present.load(Ordering::Relaxed) { + sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))") + .bind(&lane_lock_key) + .execute(tx.as_mut()) + .await + .map_err(map_sqlx_error)?; + true + } else { + sqlx::query_scalar( + r#" + SELECT COALESCE(has_function_privilege(to_regprocedure($2), 'EXECUTE'), false) + FROM (SELECT pg_advisory_xact_lock(hashtextextended($1, 0))) AS lane_lock + "#, + ) + .bind(&lane_lock_key) + .bind(format!("{schema}.provision_lane(text,smallint,smallint)")) + .fetch_one(tx.as_mut()) .await - .map_err(map_sqlx_error)?; + .map_err(map_sqlx_error)? + }; + + if provisioner_callable + && self + .provision_lane(tx, queue, priority, enqueue_shard) + .await? + { + self.mark_lane_ensured(queue, priority, enqueue_shard); + return Ok(()); + } sqlx::query(audited_sql(format!( r#" @@ -4712,6 +4745,63 @@ impl QueueStorage { Ok(()) } + /// Create the lane through `{schema}.provision_lane`, which runs as the + /// schema owner, inside a savepoint. Returns false, with the transaction + /// as it was, when the function has gone (`42883`) or the schema owner + /// may not write the lane (`42501`); the caller then writes the rows + /// itself, which is what a binary that predates the function does. + async fn provision_lane<'a>( + &self, + tx: &mut sqlx::Transaction<'a, sqlx::Postgres>, + queue: &str, + priority: i16, + enqueue_shard: i16, + ) -> Result { + let schema = self.schema(); + sqlx::query("SAVEPOINT awa_provision_lane") + .execute(tx.as_mut()) + .await + .map_err(map_sqlx_error)?; + let provisioned = sqlx::query(audited_sql(format!( + "SELECT {schema}.provision_lane($1, $2, $3)" + ))) + .bind(queue) + .bind(priority) + .bind(enqueue_shard) + .execute(tx.as_mut()) + .await; + match provisioned { + Ok(_) => { + sqlx::query("RELEASE SAVEPOINT awa_provision_lane") + .execute(tx.as_mut()) + .await + .map_err(map_sqlx_error)?; + self.lane_provisioner_present.store(true, Ordering::Relaxed); + Ok(true) + } + Err(sqlx::Error::Database(db_err)) + if matches!(db_err.code().as_deref(), Some("42883" | "42501")) => + { + sqlx::query("ROLLBACK TO SAVEPOINT awa_provision_lane") + .execute(tx.as_mut()) + .await + .map_err(map_sqlx_error)?; + self.lane_provisioner_present + .store(false, Ordering::Relaxed); + if !self.lane_provisioner_warned.swap(true, Ordering::Relaxed) { + tracing::warn!( + schema, + code = db_err.code().as_deref().unwrap_or_default(), + error = %db_err.message(), + "Lane provisioning function failed; creating lane rows as the connected role" + ); + } + Ok(false) + } + Err(err) => Err(map_sqlx_error(err)), + } + } + /// Pick the enqueue shard for a row on `queue`. /// /// Reads `awa.queue_meta.enqueue_shards` (default 1, cached diff --git a/awa-model/src/storage.rs b/awa-model/src/storage.rs index b7a0ff1ad..d67dc2048 100644 --- a/awa-model/src/storage.rs +++ b/awa-model/src/storage.rs @@ -469,3 +469,79 @@ pub async fn status_report(pool: &PgPool) -> Result, + pub reassign_sql: Option, +} + +/// What [`repair_lane_sequences`] did to one lane sequence that had a problem. +#[derive(Debug, Clone, Serialize, Deserialize, FromRow, PartialEq, Eq)] +pub struct LaneSequenceRepair { + pub sequence_name: String, + /// Owner after the repair. + pub owner: String, + pub expected_owner: String, + /// `repaired`; `owner_unchanged` when the grants are complete but the + /// connected role could not reassign the sequence; or `unrepaired` when + /// grants are still missing too. + pub outcome: String, + pub missing_grants: Vec, + pub reassign_sql: Option, +} + +/// Lane sequences in `schema` that need repair. Read-only; any role may run it. +pub async fn lane_sequence_problems<'e, E>( + executor: E, + schema: &str, +) -> Result, AwaError> +where + E: PgExecutor<'e>, +{ + Ok(sqlx::query_as( + "SELECT sequence_name, owner, expected_owner, missing_grants, reassign_sql \ + FROM awa.lane_sequence_problems($1)", + ) + .bind(schema) + .fetch_all(executor) + .await?) +} + +/// Give each lane sequence in `schema` to the schema owner and add its missing +/// grants, as far as the connected role is allowed to. A sequence the role +/// cannot reassign is returned with the statements still to run, never an +/// error. +pub async fn repair_lane_sequences( + pool: &PgPool, + schema: &str, +) -> Result, AwaError> { + Ok(sqlx::query_as( + "SELECT sequence_name, owner, expected_owner, outcome, missing_grants, reassign_sql \ + FROM awa.repair_lane_sequences($1)", + ) + .bind(schema) + .fetch_all(pool) + .await?) +} + +/// How many lane sequences need repair in each queue-storage schema that has +/// any. Read-only. +pub async fn lane_sequence_problem_counts(pool: &PgPool) -> Result, AwaError> { + Ok(sqlx::query_as( + "SELECT discovered.schema_name, count(*) \ + FROM awa.queue_storage_schemas() AS discovered \ + CROSS JOIN LATERAL awa.lane_sequence_problems(discovered.schema_name) AS problem \ + GROUP BY discovered.schema_name \ + ORDER BY discovered.schema_name", + ) + .fetch_all(pool) + .await?) +} diff --git a/awa-model/tests/least_privilege_runtime_test.rs b/awa-model/tests/least_privilege_runtime_test.rs index 99e69d536..db66046a4 100644 --- a/awa-model/tests/least_privilege_runtime_test.rs +++ b/awa-model/tests/least_privilege_runtime_test.rs @@ -250,21 +250,26 @@ async fn check_runtime(pool: &PgPool, schema: &str) -> TestResult { INSERT INTO {schema}.queue_claim_heads(queue, priority, enqueue_shard) VALUES ('prepared_queue', 2, 0) ON CONFLICT DO NOTHING; SELECT {schema}.reserve_enqueue_seq('prepared_queue', 2::smallint, 0::smallint, 3); "#)))).await?; - let error = pool - .execute(sqlx::raw_sql(awa_model::audited_sql(format!( - "SELECT {schema}.ensure_lane_sequences('unprepared_queue', 2::smallint, 0::smallint)" - )))) - .await - .unwrap_err(); - assert_eq!( - error.as_database_error().unwrap().code().as_deref(), - Some("42501") - ); - assert!(error - .as_database_error() - .unwrap() - .message() - .contains("has not been provisioned")); + // An unprepared lane is created through the schema owner's provisioning + // function, so the runtime still issues no DDL of its own. + pool.execute(sqlx::raw_sql(awa_model::audited_sql(format!( + "SELECT {schema}.ensure_lane_sequences('unprepared_queue', 2::smallint, 0::smallint); \ + SELECT {schema}.provision_lane('unprepared_queue', 2::smallint, 0::smallint);" + )))) + .await?; + let foreign_owned: i64 = sqlx::query_scalar(awa_model::audited_sql(format!( + "SELECT count(*) FROM pg_class \ + WHERE relnamespace = '{schema}'::regnamespace AND relkind = 'S' \ + AND relowner <> (SELECT nspowner FROM pg_namespace WHERE nspname = '{schema}')" + ))) + .fetch_one(pool) + .await?; + assert_eq!(foreign_owned, 0); + sqlx::query_scalar::<_, i64>(awa_model::audited_sql(format!( + "SELECT {schema}.reserve_enqueue_seq('unprepared_queue', 2::smallint, 0::smallint, 1)" + ))) + .fetch_one(pool) + .await?; let error = pool .execute(sqlx::raw_sql(awa_model::audited_sql(format!( "CREATE TABLE {schema}.must_not_create(id integer)" @@ -331,10 +336,14 @@ async fn exercise_lazy_lane(options: PgConnectOptions, owner: &str, runtime: &st "GRANT USAGE, CREATE ON SCHEMA awa TO {runtime}; \ GRANT SELECT, INSERT, UPDATE, DELETE, TRUNCATE ON ALL TABLES IN SCHEMA awa TO {runtime}; \ GRANT USAGE, SELECT, UPDATE ON ALL SEQUENCES IN SCHEMA awa TO {runtime}; \ - GRANT EXECUTE ON ALL FUNCTIONS IN SCHEMA awa TO {runtime};" + GRANT EXECUTE ON ALL FUNCTIONS IN SCHEMA awa TO {runtime}; \ + REVOKE EXECUTE ON FUNCTION awa.provision_lane_sequences(text, smallint, smallint) \ + FROM PUBLIC, {runtime};" ), ) .await?; + // Without the provisioning function the runtime creates the lane's + // sequences itself and owns them, as a 0.6 runtime with CREATE did. let runtime_pool = PgPoolOptions::new() .max_connections(2) .connect_with(options.username(runtime)) diff --git a/awa/tests/fixtures/v052_lane_sequence_helpers.sql b/awa/tests/fixtures/v052_lane_sequence_helpers.sql new file mode 100644 index 000000000..b7d396dad --- /dev/null +++ b/awa/tests/fixtures/v052_lane_sequence_helpers.sql @@ -0,0 +1,190 @@ +-- The lane-sequence helpers as migration v052 left them, before lane +-- provisioning functions existed. Installed under a separate name so a test +-- can put a schema back in its v052 shape. +CREATE OR REPLACE FUNCTION awa.v052_refresh_lane_sequence_helpers(p_schema TEXT) +RETURNS VOID +LANGUAGE plpgsql +SECURITY INVOKER +SET search_path = pg_catalog +AS $refresh$ +BEGIN + EXECUTE format( + $ddl$ + CREATE OR REPLACE FUNCTION %1$I.ensure_lane_sequences( + p_queue TEXT, + p_priority SMALLINT, + p_enqueue_shard SMALLINT + ) + RETURNS VOID + LANGUAGE plpgsql + SET search_path = pg_catalog + AS $func$ + DECLARE + v_enqueue_seq TEXT := %1$I.queue_lane_sequence_name( + 'queue_enqueue_seq', + p_queue, + p_priority, + p_enqueue_shard + ); + v_claim_seq TEXT := %1$I.queue_lane_sequence_name( + 'queue_claim_seq', + p_queue, + p_priority, + p_enqueue_shard + ); + BEGIN + IF to_regclass(format('%%I.%%I', %1$L, v_enqueue_seq)) IS NULL THEN + IF NOT has_schema_privilege(current_user, %1$L, 'CREATE') THEN + RAISE EXCEPTION 'lane sequence %%.%% has not been provisioned', %1$L, v_enqueue_seq + USING ERRCODE = '42501', + HINT = 'Run awa storage prepare-queue as the migrator before starting producers or workers.'; + END IF; + EXECUTE format( + 'CREATE SEQUENCE IF NOT EXISTS %%I.%%I AS bigint START WITH 1 MINVALUE 1 CACHE 1', + %1$L, + v_enqueue_seq + ); + END IF; + IF to_regclass(format('%%I.%%I', %1$L, v_claim_seq)) IS NULL THEN + IF NOT has_schema_privilege(current_user, %1$L, 'CREATE') THEN + RAISE EXCEPTION 'lane sequence %%.%% has not been provisioned', %1$L, v_claim_seq + USING ERRCODE = '42501', + HINT = 'Run awa storage prepare-queue as the migrator before starting producers or workers.'; + END IF; + EXECUTE format( + 'CREATE SEQUENCE IF NOT EXISTS %%I.%%I AS bigint START WITH 1 MINVALUE 1 CACHE 1', + %1$L, + v_claim_seq + ); + END IF; + + UPDATE %1$I.queue_enqueue_heads + SET seq_name = v_enqueue_seq + WHERE queue = p_queue + AND priority = p_priority + AND enqueue_shard = p_enqueue_shard + AND seq_name IS DISTINCT FROM v_enqueue_seq; + + UPDATE %1$I.queue_claim_heads + SET seq_name = v_claim_seq + WHERE queue = p_queue + AND priority = p_priority + AND enqueue_shard = p_enqueue_shard + AND seq_name IS DISTINCT FROM v_claim_seq; + END; + $func$ + $ddl$, + p_schema + ); + + EXECUTE format( + $ddl$ + CREATE OR REPLACE FUNCTION %1$I.queue_enqueue_head_sequence_sync() + RETURNS TRIGGER + LANGUAGE plpgsql + SET search_path = pg_catalog + AS $func$ + DECLARE + v_seq_name TEXT := %1$I.queue_lane_sequence_name( + 'queue_enqueue_seq', + NEW.queue, + NEW.priority, + NEW.enqueue_shard + ); + v_qualified_seq TEXT := format('%%I.%%I', %1$L, v_seq_name); + v_count BIGINT; + v_start BIGINT; + BEGIN + IF to_regclass(format('%%I.%%I', %1$L, v_seq_name)) IS NULL THEN + IF NOT has_schema_privilege(current_user, %1$L, 'CREATE') THEN + RAISE EXCEPTION 'lane sequence %%.%% has not been provisioned', %1$L, v_seq_name + USING ERRCODE = '42501', + HINT = 'Run awa storage prepare-queue as the migrator before starting producers or workers.'; + END IF; + EXECUTE format( + 'CREATE SEQUENCE IF NOT EXISTS %%I.%%I AS bigint START WITH 1 MINVALUE 1 CACHE 1', + %1$L, + v_seq_name + ); + END IF; + NEW.seq_name := v_seq_name; + + IF TG_OP = 'UPDATE' + AND NEW.next_seq IS DISTINCT FROM OLD.next_seq + AND NEW.next_seq > OLD.next_seq + THEN + v_count := NEW.next_seq - OLD.next_seq; + PERFORM pg_advisory_lock(hashtextextended(v_qualified_seq, 0)); + BEGIN + EXECUTE format( + 'SELECT nextval(%%L::regclass)::bigint', + v_qualified_seq + ) + INTO v_start; + + IF v_count > 1 THEN + EXECUTE format( + 'SELECT setval(%%L::regclass, %%s, true)', + v_qualified_seq, + v_start + v_count - 1 + ); + END IF; + EXCEPTION WHEN OTHERS THEN + PERFORM pg_advisory_unlock(hashtextextended(v_qualified_seq, 0)); + RAISE; + END; + PERFORM pg_advisory_unlock(hashtextextended(v_qualified_seq, 0)); + NEW.next_seq := v_start + v_count; + ELSE + PERFORM %1$I.set_sequence_next(v_seq_name, NEW.next_seq); + END IF; + + RETURN NEW; + END; + $func$ + $ddl$, + p_schema + ); + + EXECUTE format( + $ddl$ + CREATE OR REPLACE FUNCTION %1$I.queue_claim_head_sequence_sync() + RETURNS TRIGGER + LANGUAGE plpgsql + SET search_path = pg_catalog + AS $func$ + DECLARE + v_seq_name TEXT := %1$I.queue_lane_sequence_name( + 'queue_claim_seq', + NEW.queue, + NEW.priority, + NEW.enqueue_shard + ); + BEGIN + IF to_regclass(format('%%I.%%I', %1$L, v_seq_name)) IS NULL THEN + IF NOT has_schema_privilege(current_user, %1$L, 'CREATE') THEN + RAISE EXCEPTION 'lane sequence %%.%% has not been provisioned', %1$L, v_seq_name + USING ERRCODE = '42501', + HINT = 'Run awa storage prepare-queue as the migrator before starting producers or workers.'; + END IF; + EXECUTE format( + 'CREATE SEQUENCE IF NOT EXISTS %%I.%%I AS bigint START WITH 1 MINVALUE 1 CACHE 1', + %1$L, + v_seq_name + ); + END IF; + NEW.seq_name := v_seq_name; + + IF TG_OP = 'INSERT' THEN + PERFORM %1$I.set_sequence_next(v_seq_name, NEW.claim_seq); + END IF; + + RETURN NEW; + END; + $func$ + $ddl$, + p_schema + ); + +END; +$refresh$; diff --git a/awa/tests/lane_ownership_roles_test.rs b/awa/tests/lane_ownership_roles_test.rs new file mode 100644 index 000000000..e5aadbd39 --- /dev/null +++ b/awa/tests/lane_ownership_roles_test.rs @@ -0,0 +1,735 @@ +//! Queue lanes under a split-role install: a schema owner, a migration role +//! that is a member of it, and several application roles that share a runtime +//! group role. Only the owner and the migration role have default privileges, +//! so a lane sequence is usable by every runtime role only if it is created as +//! the owner or granted explicitly, whichever role touches the lane first. + +use awa::model::{migrations, storage, AwaError, InsertOpts, InsertParams, QueueStorage}; +use awa::model::{JobState, QueueStorageConfig}; +use sqlx::postgres::{PgConnectOptions, PgPoolOptions}; +use sqlx::{Executor, PgPool}; +use std::str::FromStr; +use std::time::Duration; + +type TestResult = Result<(), Box>; + +fn sql(text: impl Into) -> sqlx::AssertSqlSafe { + awa::audited_sql(text) +} + +/// Roles, a database owned by the schema owner, and a pool per role. +struct Cluster { + admin: PgPool, + database: String, + roles: Vec, + owner_name: String, + migrator_name: String, + runtime_name: String, + owner: PgPool, + migrator: PgPool, + app_a: PgPool, + app_b: PgPool, + app_c: PgPool, +} + +#[derive(Clone, Copy, PartialEq)] +enum Schema { + /// Every migration, including the lane provisioning functions. + Current, + /// The release before lane provisioning (v052). + BeforeProvisioning, +} + +impl Cluster { + /// `runtime_create` grants the runtime group CREATE on the schema, as a + /// 0.6 install does so runtimes can create lane sequences themselves. + async fn new( + schema: Schema, + runtime_create: bool, + ) -> Result> { + let url = std::env::var("DATABASE_URL") + .unwrap_or_else(|_| "postgres://postgres:test@localhost:15432/awa_test".to_string()); + let admin = PgPoolOptions::new() + .max_connections(2) + .connect(&url) + .await?; + let suffix = uuid::Uuid::new_v4().simple().to_string()[..12].to_string(); + let owner_name = format!("lane_owner_{suffix}"); + let migrator_name = format!("lane_migrator_{suffix}"); + let runtime_name = format!("lane_runtime_{suffix}"); + let app_names = ["a", "b", "c"].map(|app| format!("lane_app_{app}_{suffix}")); + let database = format!("lane_roles_{suffix}"); + admin + .execute(sqlx::raw_sql(sql(format!( + r#" + CREATE ROLE {owner_name} LOGIN PASSWORD 'lane_test'; + CREATE ROLE {migrator_name} LOGIN PASSWORD 'lane_test' IN ROLE {owner_name}; + CREATE ROLE {runtime_name} NOLOGIN; + CREATE ROLE {a} LOGIN PASSWORD 'lane_test' IN ROLE {runtime_name}; + CREATE ROLE {b} LOGIN PASSWORD 'lane_test' IN ROLE {runtime_name}; + CREATE ROLE {c} LOGIN PASSWORD 'lane_test' IN ROLE {runtime_name}; + "#, + a = app_names[0], + b = app_names[1], + c = app_names[2], + )))) + .await?; + admin + .execute(sqlx::raw_sql(sql(format!( + "CREATE DATABASE {database} OWNER {owner_name}" + )))) + .await?; + let mut roles = vec![ + owner_name.clone(), + migrator_name.clone(), + runtime_name.clone(), + ]; + roles.extend(app_names.iter().cloned()); + let options = PgConnectOptions::from_str(&url)? + .database(&database) + .password("lane_test"); + let pool = |user: &str| { + PgPoolOptions::new() + .max_connections(3) + .connect_with(options.clone().username(user)) + }; + let owner = pool(&owner_name).await?; + let migrator = pool(&migrator_name).await?; + + owner + .execute(sqlx::raw_sql(sql(format!( + r#" + CREATE SCHEMA awa; + GRANT USAGE ON SCHEMA awa TO {runtime_name}; + "# + )))) + .await?; + for (creator, creator_pool) in [(&owner_name, &owner), (&migrator_name, &migrator)] { + creator_pool + .execute(sqlx::raw_sql(sql(format!( + r#" + ALTER DEFAULT PRIVILEGES FOR ROLE {creator} IN SCHEMA awa + GRANT SELECT, INSERT, UPDATE, DELETE, TRUNCATE ON TABLES TO {runtime_name}; + ALTER DEFAULT PRIVILEGES FOR ROLE {creator} IN SCHEMA awa + GRANT USAGE, SELECT, UPDATE ON SEQUENCES TO {runtime_name}; + ALTER DEFAULT PRIVILEGES FOR ROLE {creator} IN SCHEMA awa + GRANT EXECUTE ON FUNCTIONS TO {runtime_name}; + "# + )))) + .await?; + } + // The migration role runs every migration with the owner as its + // effective role, as the database-roles guide describes. + let migrate_as_owner = PgPoolOptions::new() + .max_connections(1) + .connect_with( + options + .clone() + .username(&migrator_name) + .options([("role", owner_name.as_str())]), + ) + .await?; + let target = match schema { + Schema::Current => migrations::CURRENT_VERSION, + Schema::BeforeProvisioning => 52, + }; + for (_, _, statement) in migrations::migration_sql_range(0, target) { + migrate_as_owner + .execute(sqlx::raw_sql(sql(statement))) + .await?; + } + if schema == Schema::BeforeProvisioning { + // Every migration bundle carries the current helper definitions, + // so put the v052 helpers back and drop what v052 did not have. + migrate_as_owner + .execute(sqlx::raw_sql(sql(include_str!( + "fixtures/v052_lane_sequence_helpers.sql" + )))) + .await?; + migrate_as_owner + .execute(sqlx::raw_sql(sql(r#" + SELECT awa.v052_refresh_lane_sequence_helpers('awa'); + DROP FUNCTION awa.v052_refresh_lane_sequence_helpers(text); + DROP FUNCTION awa.provision_lane(text, smallint, smallint); + DROP FUNCTION awa.provision_lane_sequences(text, smallint, smallint); + DROP FUNCTION awa.repair_lane_sequences(text); + DROP FUNCTION awa.lane_sequence_problems(text, text); + "#))) + .await?; + } + migrate_as_owner.close().await; + if runtime_create { + owner + .execute(sqlx::raw_sql(sql(format!( + "GRANT CREATE ON SCHEMA awa TO {runtime_name}" + )))) + .await?; + } + + let app_a = pool(&app_names[0]).await?; + let app_b = pool(&app_names[1]).await?; + let app_c = pool(&app_names[2]).await?; + Ok(Self { + admin, + database, + roles, + owner_name, + migrator_name, + runtime_name, + owner, + migrator, + app_a, + app_b, + app_c, + }) + } + + async fn teardown(self) -> TestResult { + for pool in [ + &self.owner, + &self.migrator, + &self.app_a, + &self.app_b, + &self.app_c, + ] { + pool.close().await; + } + self.admin + .execute(sqlx::raw_sql(sql(format!( + "DROP DATABASE {} WITH (FORCE)", + self.database + )))) + .await?; + for role in self.roles.iter().rev() { + self.admin + .execute(sqlx::raw_sql(sql(format!("DROP ROLE {role}")))) + .await?; + } + self.admin.close().await; + Ok(()) + } + + /// `(sequence, owner)` for every lane sequence of `queue`. + async fn lane_sequence_owners( + &self, + queue: &str, + ) -> Result, sqlx::Error> { + sqlx::query_as( + r#" + SELECT sequence.relname::text, pg_get_userbyid(sequence.relowner)::text + FROM awa.queue_enqueue_heads AS heads + CROSS JOIN LATERAL (VALUES (heads.seq_name), (( + SELECT claims.seq_name FROM awa.queue_claim_heads AS claims + WHERE claims.queue = heads.queue + AND claims.priority = heads.priority + AND claims.enqueue_shard = heads.enqueue_shard + ))) AS lane(seq_name) + JOIN pg_class AS sequence + ON sequence.relname = lane.seq_name + AND sequence.relnamespace = 'awa'::regnamespace + WHERE heads.queue = $1 + ORDER BY 1 + "#, + ) + .bind(queue) + .fetch_all(&self.owner) + .await + } + + async fn assert_lanes_owned_by_owner(&self, queue: &str, lanes: usize) -> TestResult { + let owners = self.lane_sequence_owners(queue).await?; + assert_eq!( + owners.len(), + lanes * 2, + "lane sequences of {queue}: {owners:?}" + ); + for (sequence, owner) in owners { + assert_eq!(owner, self.owner_name, "{sequence} of {queue}"); + } + Ok(()) + } + + async fn set_enqueue_shards(&self, queue: &str, shards: i16) -> Result<(), sqlx::Error> { + sqlx::query( + "INSERT INTO awa.queue_meta (queue, enqueue_shards) VALUES ($1, $2) \ + ON CONFLICT (queue) DO UPDATE SET enqueue_shards = EXCLUDED.enqueue_shards", + ) + .bind(queue) + .bind(shards) + .execute(&self.owner) + .await?; + Ok(()) + } +} + +fn store() -> QueueStorage { + QueueStorage::new(QueueStorageConfig::default()).expect("default queue storage config") +} + +/// Enqueue `count` jobs through a fresh store, as a producer process would. +async fn enqueue(pool: &PgPool, queue: &str, priority: i16, count: i64) -> Result { + store().enqueue_batch(pool, queue, priority, count).await +} + +/// Claim and complete everything on `queue`, as a worker does; returns how +/// many jobs completed. +async fn drain(pool: &PgPool, queue: &str) -> Result { + let worker = store(); + let mut completed = 0; + loop { + let claimed = worker + .claim_runtime_batch(pool, queue, 100, Duration::from_secs(60)) + .await?; + if claimed.is_empty() { + return Ok(completed); + } + completed += worker.complete_runtime_batch(pool, &claimed).await?.len(); + } +} + +/// Schedule jobs that become due shortly, one per priority. +async fn schedule(pool: &PgPool, queue: &str, priorities: &[i16]) -> Result<(), AwaError> { + let run_at = chrono::Utc::now() + chrono::Duration::milliseconds(200); + let jobs: Vec = priorities + .iter() + .map(|&priority| InsertParams { + kind: "lane_ownership_job".into(), + args: serde_json::json!({}), + opts: InsertOpts { + queue: queue.into(), + priority, + run_at: Some(run_at), + ..Default::default() + }, + }) + .collect(); + store().enqueue_params_batch(pool, &jobs).await?; + tokio::time::sleep(Duration::from_millis(300)).await; + Ok(()) +} + +async fn first_touch_by_one_app_role(runtime_create: bool) -> TestResult { + let cluster = Cluster::new(Schema::Current, runtime_create).await?; + + // Role A creates the lane; role B enqueues, claims and completes on it. + assert_eq!(enqueue(&cluster.app_a, "fresh_queue", 3, 2).await?, 2); + assert_eq!(enqueue(&cluster.app_b, "fresh_queue", 3, 2).await?, 2); + assert_eq!(drain(&cluster.app_b, "fresh_queue").await?, 4); + + // The maintenance leader, running as role C, promotes scheduled jobs into + // lanes nobody has created yet: two priorities over two shards. + cluster.set_enqueue_shards("fresh_queue", 2).await?; + schedule(&cluster.app_a, "fresh_queue", &[1, 4]).await?; + assert_eq!( + store() + .promote_due(&cluster.app_c, JobState::Scheduled, 10) + .await?, + 2 + ); + let promoted_shards: Vec = sqlx::query_scalar( + "SELECT enqueue_shard FROM awa.queue_enqueue_heads \ + WHERE queue = 'fresh_queue' AND priority IN (1, 4) ORDER BY 1", + ) + .fetch_all(&cluster.owner) + .await?; + assert_eq!(promoted_shards, vec![0, 1]); + + // A and B both use the lanes the leader created; each producer's rotor + // visits both shards. + let (producer_a, producer_b) = (store(), store()); + for _ in 0..2 { + producer_a + .enqueue_batch(&cluster.app_a, "fresh_queue", 4, 1) + .await?; + producer_b + .enqueue_batch(&cluster.app_b, "fresh_queue", 1, 1) + .await?; + } + assert_eq!(drain(&cluster.app_a, "fresh_queue").await?, 6); + + cluster + .assert_lanes_owned_by_owner("fresh_queue", 5) + .await?; + assert!(storage::lane_sequence_problems(&cluster.owner, "awa") + .await? + .is_empty()); + cluster.teardown().await +} + +#[tokio::test] +async fn lane_first_touched_by_an_app_role_with_schema_create_serves_every_runtime_role( +) -> TestResult { + first_touch_by_one_app_role(true).await +} + +#[tokio::test] +async fn lane_first_touched_by_an_app_role_without_schema_create_serves_every_runtime_role( +) -> TestResult { + first_touch_by_one_app_role(false).await +} + +#[tokio::test] +async fn raising_enqueue_shards_live_lets_the_leader_create_a_shard_first() -> TestResult { + let cluster = Cluster::new(Schema::Current, true).await?; + enqueue(&cluster.app_a, "sharded_queue", 2, 3).await?; + + cluster.set_enqueue_shards("sharded_queue", 2).await?; + // The leader's store has already routed a promotion to shard 0, as a + // long-running leader has, so its next pick is the new shard. + let leader = store(); + schedule(&cluster.app_a, "sharded_queue", &[2]).await?; + assert_eq!( + leader + .promote_due(&cluster.app_c, JobState::Scheduled, 10) + .await?, + 1 + ); + schedule(&cluster.app_b, "sharded_queue", &[2]).await?; + assert_eq!( + leader + .promote_due(&cluster.app_c, JobState::Scheduled, 10) + .await?, + 1 + ); + let shards: Vec = sqlx::query_scalar( + "SELECT enqueue_shard FROM awa.queue_enqueue_heads WHERE queue = 'sharded_queue' ORDER BY 1", + ) + .fetch_all(&cluster.owner) + .await?; + assert_eq!(shards, vec![0, 1]); + + // Producers on both roles spread over both shards; B drains them. + let producer_a = store(); + let producer_b = store(); + for _ in 0..4 { + producer_a + .enqueue_batch(&cluster.app_a, "sharded_queue", 2, 1) + .await?; + producer_b + .enqueue_batch(&cluster.app_b, "sharded_queue", 2, 1) + .await?; + } + assert_eq!(drain(&cluster.app_b, "sharded_queue").await?, 13); + cluster + .assert_lanes_owned_by_owner("sharded_queue", 2) + .await?; + cluster.teardown().await +} + +#[tokio::test] +async fn prepare_queue_as_an_app_role_provisions_lanes_every_runtime_role_can_use() -> TestResult { + let cluster = Cluster::new(Schema::Current, true).await?; + store() + .prepare_queue(&cluster.app_a, "prepared_queue", 2) + .await?; + cluster + .assert_lanes_owned_by_owner("prepared_queue", 8) + .await?; + + cluster.set_enqueue_shards("prepared_queue", 2).await?; + let producer = store(); + for priority in 1..=4 { + producer + .enqueue_batch(&cluster.app_b, "prepared_queue", priority, 1) + .await?; + producer + .enqueue_batch(&cluster.app_c, "prepared_queue", priority, 1) + .await?; + } + assert_eq!(drain(&cluster.app_b, "prepared_queue").await?, 8); + cluster + .assert_lanes_owned_by_owner("prepared_queue", 8) + .await?; + cluster.teardown().await +} + +#[tokio::test] +async fn mixed_fleet_shares_lanes_whichever_path_creates_them() -> TestResult { + let cluster = Cluster::new(Schema::Current, true).await?; + // Role A cannot call provision_lane, so its stores take the path a binary + // that predates the function takes: they write the lane rows themselves. + cluster + .owner + .execute(sqlx::raw_sql(sql(format!( + "REVOKE EXECUTE ON FUNCTION awa.provision_lane(text, smallint, smallint) FROM PUBLIC, {}", + cluster.runtime_name + )))) + .await?; + cluster + .owner + .execute(sqlx::raw_sql(sql(format!( + "GRANT EXECUTE ON FUNCTION awa.provision_lane(text, smallint, smallint) TO {}, {}", + cluster.roles[4], cluster.roles[5] + )))) + .await?; + let can_provision: bool = sqlx::query_scalar( + "SELECT has_function_privilege('awa.provision_lane(text,smallint,smallint)', 'EXECUTE')", + ) + .fetch_one(&cluster.app_a) + .await?; + assert!(!can_provision); + + enqueue(&cluster.app_a, "fleet_queue", 1, 2).await?; + enqueue(&cluster.app_b, "fleet_queue", 2, 2).await?; + enqueue(&cluster.app_b, "fleet_queue", 1, 2).await?; + enqueue(&cluster.app_a, "fleet_queue", 2, 2).await?; + assert_eq!(drain(&cluster.app_a, "fleet_queue").await?, 8); + cluster + .assert_lanes_owned_by_owner("fleet_queue", 2) + .await?; + + // A provisioner that disappears after a store has used it: the next lane + // falls back inside the same transaction instead of failing the enqueue. + let producer = store(); + producer + .enqueue_batch(&cluster.app_b, "fleet_queue", 3, 1) + .await?; + cluster + .owner + .execute("DROP FUNCTION awa.provision_lane(text, smallint, smallint)") + .await?; + producer + .enqueue_batch(&cluster.app_b, "fleet_queue", 4, 1) + .await?; + enqueue(&cluster.app_c, "fleet_queue", 4, 1).await?; + assert_eq!(drain(&cluster.app_c, "fleet_queue").await?, 3); + cluster + .assert_lanes_owned_by_owner("fleet_queue", 4) + .await?; + cluster.teardown().await +} + +#[tokio::test] +async fn binary_ahead_of_the_schema_creates_lanes_as_before() -> TestResult { + let cluster = Cluster::new(Schema::BeforeProvisioning, false).await?; + let has_provisioner: bool = sqlx::query_scalar( + "SELECT to_regprocedure('awa.provision_lane(text,smallint,smallint)') IS NOT NULL", + ) + .fetch_one(&cluster.owner) + .await?; + assert!(!has_provisioner); + + // The owner creates lanes lazily, as on v052. + enqueue(&cluster.owner, "ahead_queue", 2, 2).await?; + enqueue(&cluster.app_b, "ahead_queue", 2, 1).await?; + assert_eq!(drain(&cluster.app_b, "ahead_queue").await?, 3); + // A runtime without schema CREATE still gets v052's provisioning hint. + let error = enqueue(&cluster.app_a, "ahead_queue", 3, 1) + .await + .unwrap_err(); + let AwaError::Database(sqlx::Error::Database(db_error)) = &error else { + panic!("expected a database error, got {error:?}"); + }; + assert_eq!(db_error.code().as_deref(), Some("42501")); + assert!(db_error.message().contains("has not been provisioned")); + cluster.teardown().await +} + +#[tokio::test] +async fn concurrent_first_touch_of_one_lane_by_two_roles() -> TestResult { + let cluster = Cluster::new(Schema::Current, true).await?; + + // A holds a brand-new lane's creation open; B's first touch waits for it + // and then uses A's sequences. + let mut holding = cluster.app_a.begin().await?; + sqlx::query("SELECT awa.provision_lane('raced_queue', 2::smallint, 0::smallint)") + .execute(holding.as_mut()) + .await?; + let app_b = cluster.app_b.clone(); + let waiting = tokio::spawn(async move { enqueue(&app_b, "raced_queue", 2, 1).await }); + let mut blocked = false; + for _ in 0..100 { + blocked = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM pg_locks WHERE locktype = 'advisory' AND NOT granted \ + AND database = (SELECT oid FROM pg_database WHERE datname = current_database()))", + ) + .fetch_one(&cluster.owner) + .await?; + if blocked { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert!(blocked, "B's first touch should wait on A's lane lock"); + holding.commit().await?; + assert_eq!(waiting.await??, 1); + + // Unordered races on many fresh lanes. + for round in 0..20 { + let queue = format!("raced_queue_{round}"); + let (from_a, from_b) = tokio::join!( + enqueue(&cluster.app_a, &queue, 1, 1), + enqueue(&cluster.app_b, &queue, 1, 1) + ); + assert_eq!(from_a? + from_b?, 2); + cluster.assert_lanes_owned_by_owner(&queue, 1).await?; + assert_eq!(drain(&cluster.app_c, &queue).await?, 2); + } + assert_eq!(drain(&cluster.app_c, "raced_queue").await?, 1); + cluster + .assert_lanes_owned_by_owner("raced_queue", 1) + .await?; + cluster.teardown().await +} + +#[tokio::test] +async fn migration_repairs_lanes_it_may_and_reports_the_rest() -> TestResult { + let mut cluster = Cluster::new(Schema::BeforeProvisioning, true).await?; + // Lanes as a 0.6 fleet left them: one each created by an app role, by the + // migration role acting as itself, and by the owner without the runtime's + // grant. + enqueue(&cluster.app_a, "app_lane", 1, 1).await?; + enqueue(&cluster.migrator, "migrator_lane", 1, 1).await?; + enqueue(&cluster.owner, "owner_lane", 1, 1).await?; + for (sequence, _) in cluster.lane_sequence_owners("owner_lane").await? { + cluster + .owner + .execute(sqlx::raw_sql(sql(format!( + "REVOKE ALL ON SEQUENCE awa.{sequence} FROM {}", + cluster.runtime_name + )))) + .await?; + } + let failed = drain(&cluster.app_b, "owner_lane").await.unwrap_err(); + assert!( + failed + .to_string() + .contains("permission denied for sequence"), + "{failed}" + ); + let failed = drain(&cluster.app_b, "app_lane").await.unwrap_err(); + assert!( + failed + .to_string() + .contains("permission denied for sequence"), + "{failed}" + ); + // 0.7 refuses to migrate while app roles hold CREATE on the schema. + cluster + .owner + .execute(sqlx::raw_sql(sql(format!( + "REVOKE CREATE ON SCHEMA awa FROM {}", + cluster.runtime_name + )))) + .await?; + + // The migration runs as the migration role itself. It is a member of the + // owner, so it may reassign its own and the owner's sequences, but not + // the app role's. + for (_, _, statement) in migrations::migration_sql_range(52, migrations::CURRENT_VERSION) { + cluster + .migrator + .execute(sqlx::raw_sql(sql(statement))) + .await?; + } + assert_eq!( + migrations::current_version(&cluster.owner).await?, + migrations::CURRENT_VERSION + ); + cluster + .assert_lanes_owned_by_owner("migrator_lane", 1) + .await?; + cluster.assert_lanes_owned_by_owner("owner_lane", 1).await?; + assert_eq!(drain(&cluster.app_b, "migrator_lane").await?, 1); + assert_eq!(drain(&cluster.app_b, "owner_lane").await?, 1); + + let problems = storage::lane_sequence_problems(&cluster.app_c, "awa").await?; + assert_eq!(problems.len(), 2, "{problems:?}"); + let app_a = cluster.roles[3].clone(); + let app_a = &app_a; + for problem in &problems { + assert_eq!(&problem.owner, app_a); + assert_eq!(problem.expected_owner, cluster.owner_name); + assert_eq!( + problem.missing_grants, + vec![format!( + "GRANT USAGE, UPDATE, SELECT ON SEQUENCE awa.{} TO {}", + problem.sequence_name, cluster.runtime_name + )] + ); + assert_eq!( + problem.reassign_sql.as_deref(), + Some( + format!( + "ALTER SEQUENCE awa.{} OWNER TO {}", + problem.sequence_name, cluster.owner_name + ) + .as_str() + ) + ); + } + + // Re-running the repair as the migration role changes nothing and fails + // nothing. The migration role created the repair function, so it grants + // its use. + cluster + .migrator + .execute(sqlx::raw_sql(sql(format!( + "GRANT EXECUTE ON FUNCTION awa.repair_lane_sequences(text) TO {}", + cluster.runtime_name + )))) + .await?; + let rerun = storage::repair_lane_sequences(&cluster.migrator, "awa").await?; + assert_eq!(rerun.len(), 2); + assert!(rerun.iter().all(|result| result.outcome == "unrepaired")); + + // A role that can act as the app role but cannot become the owner adds + // the grants and leaves the owner for someone who can. + let delegate = format!("{}_delegate", cluster.runtime_name); + cluster + .admin + .execute(sqlx::raw_sql(sql(format!( + "CREATE ROLE {delegate} LOGIN PASSWORD 'lane_test' IN ROLE {app_a}; \ + GRANT CONNECT ON DATABASE {} TO {delegate}", + cluster.database + )))) + .await?; + let delegate_pool = PgPoolOptions::new() + .max_connections(1) + .connect_with( + PgConnectOptions::from_str(&std::env::var("DATABASE_URL").unwrap_or_else(|_| { + "postgres://postgres:test@localhost:15432/awa_test".to_string() + }))? + .database(&cluster.database) + .username(&delegate) + .password("lane_test"), + ) + .await?; + let delegated = storage::repair_lane_sequences(&delegate_pool, "awa").await?; + delegate_pool.close().await; + assert_eq!(delegated.len(), 2); + for result in &delegated { + assert_eq!(result.outcome, "owner_unchanged"); + assert_eq!(&result.owner, app_a); + assert!(result.missing_grants.is_empty()); + assert!(result.reassign_sql.is_some()); + } + assert_eq!(drain(&cluster.app_b, "app_lane").await?, 1); + enqueue(&cluster.app_c, "app_lane", 1, 1).await?; + + // A role that is a member of both the app role and the owner reassigns. + cluster + .admin + .execute(sqlx::raw_sql(sql(format!( + "GRANT {app_a} TO {migrator}", + migrator = cluster.migrator_name + )))) + .await?; + let repaired = storage::repair_lane_sequences(&cluster.migrator, "awa").await?; + assert_eq!(repaired.len(), 2); + assert!( + repaired.iter().all(|result| result.outcome == "repaired"), + "{repaired:?}" + ); + assert!(storage::lane_sequence_problems(&cluster.app_c, "awa") + .await? + .is_empty()); + assert!(storage::repair_lane_sequences(&cluster.migrator, "awa") + .await? + .is_empty()); + cluster.assert_lanes_owned_by_owner("app_lane", 1).await?; + assert_eq!(drain(&cluster.app_b, "app_lane").await?, 1); + + cluster.roles.push(delegate); + cluster.teardown().await +} diff --git a/docs/queue-storage-substrate.md b/docs/queue-storage-substrate.md index 4476712e4..9542e3914 100644 --- a/docs/queue-storage-substrate.md +++ b/docs/queue-storage-substrate.md @@ -40,7 +40,7 @@ awa storage prepare-queue-storage-schema \ The command invokes the idempotent `awa.install_queue_storage_substrate(...)` helper under a per-schema advisory transaction lock. The helper is activation-neutral: preparing a schema does not route work to it. Activation is a separate staged transition described in [Upgrading from 0.5 to 0.6](upgrade-0.5-to-0.6.md). -The installer is `SECURITY INVOKER`; its caller needs DDL privileges on the target schema. After the migrator provisions each queue with `awa storage prepare-queue`, workers need runtime DML, sequence `USAGE`/`SELECT`/`UPDATE`, and `TRUNCATE` for guarded ring reclamation. They do not need schema CREATE. See [Database roles](security/database-roles.md). +The installer is `SECURITY INVOKER`; its caller needs DDL privileges on the target schema. Workers need runtime DML, sequence `USAGE`/`SELECT`/`UPDATE`, and `TRUNCATE` for guarded ring reclamation. They do not need schema CREATE: lanes are created through the schema owner's provisioning functions. See [Database roles](security/database-roles.md). ## Operator rules diff --git a/docs/security/database-roles.md b/docs/security/database-roles.md index b6c1557e9..1b5e0091f 100644 --- a/docs/security/database-roles.md +++ b/docs/security/database-roles.md @@ -63,36 +63,44 @@ ALTER DEFAULT PRIVILEGES FOR ROLE awa_owner IN SCHEMA awa If an existing installation created objects as `awa_migrator` without first setting the owner role, transfer every existing schema object to `awa_owner` before relying on owner-scoped defaults. New migrations should use the effective-role command above. -## Provision queues before starting restricted runtimes +## Lane sequences -From schema v049, runtime calls use existing lane sequences without issuing -`CREATE SEQUENCE`. Run provisioning as the migrator for every queue before -starting its producers or workers: +Each queue lane, `(queue, priority, enqueue_shard)`, has an enqueue and a claim sequence. From schema v053 they are created by the schema owner, whichever role first enqueues into, claims from or promotes into the lane: the lane helpers call `provision_lane` and `provision_lane_sequences` in the queue-storage schema. These are `SECURITY DEFINER` functions owned by the schema owner with a pinned `search_path`. Like the other runtime-callable functions they are executable by `PUBLIC`, so schema `USAGE` is what gates them, and they only create lanes, after validating the queue, priority (1–4) and shard (0–63). A runtime therefore needs neither schema `CREATE` nor default privileges to start using a new queue, priority or shard. + +A new lane sequence gets the privileges of the head table it backs. A role that may `INSERT` or `UPDATE` `queue_enqueue_heads` gets `USAGE, SELECT, UPDATE` on the enqueue sequences, and one that may only `SELECT` it gets `SELECT`; `queue_claim_heads` does the same for the claim sequences. Granting the runtime the table privileges above is enough, and no sequence default privileges are needed for lanes. + +Run migrations with the schema owner as the effective role, as above, so the provisioning functions belong to it. A migrator that cannot `SET ROLE` to the owner leaves them owned by itself and the migration warns. + +`awa storage prepare-queue` still creates every priority and shard of a queue in one transaction, and any runtime role may run it. Use it to create lanes before a deploy rather than on first use. + +### Check and repair lane ownership + +Lanes created before v053 belong to whichever role touched them first, and only that role, or the roles its default privileges named, can use them. A different runtime role then fails with `permission denied for sequence` on that lane. List lane sequences that are not owned by the schema owner or lack a runtime grant (read-only, any role): + +```bash +awa --database-url "$AWA_DATABASE_URL" storage lane-sequences --schema awa +``` + +```sql +SELECT * FROM awa.lane_sequence_problems('awa'); +``` + +The command exits 2 while any remain. Migration v053 runs the repair as the migrating role: for each sequence it changes the owner to the schema owner and adds the missing grants, wherever that role is allowed to. A sequence it is not allowed to change does not fail the migration. It is reported in a warning, and `awa migrate` prints a summary. Run the repair again at any time; it is idempotent: ```bash PGOPTIONS='-c role=awa_owner' \ - awa --database-url "$AWA_MIGRATOR_DATABASE_URL" storage prepare-queue \ - --queue email --enqueue-shards 1 + awa --database-url "$AWA_MIGRATOR_DATABASE_URL" storage lane-sequences --repair ``` -The command provisions all four priorities in one transaction and is safe to -repeat on a live queue: existing sequence cursors keep their positions. It does -not change routing or the queue's configured shard count. Prepare additional -shards before increasing `enqueue_shards`; match `--queue-stripe-count` to the -runtime's queue-storage configuration. With `PartitionedQueue`, provision each -physical queue. Use `--schema` for a custom storage schema. - -Apply the sequence grants above after provisioning, or set the owner's default -privileges first. Lane cursors need `UPDATE` for `setval`, in addition to `USAGE` -and `SELECT`. `CREATE SEQUENCE IF NOT EXISTS` requires schema CREATE even when -the sequence already exists, so provisioning alone does not repair pre-v049 -helpers: apply the migration first. Schema owners retain lazy creation for -single-role installations; a restricted runtime encountering an unprepared -lane fails with a provisioning hint. +Changing a sequence's owner requires a role that can `SET ROLE` to both the current owner and the schema owner. A migrator usually can for sequences it or the owner created, but not for one an unrelated application role created, and managed PostgreSQL offers no superuser. For each sequence that remains, the output gives the statements to run: + +- `missing_grants`: run them as the current owner, or as a role that can `SET ROLE` to it. Every runtime role can then use the lane, although the owner stays unchanged. +- `reassign_sql`: run it as a role that is a member of both roles, for example after `GRANT app_role TO awa_migrator` from a role with `ADMIN` on `app_role`. Revoke that membership afterwards. +## Custom queue-storage schemas ## Custom queue-storage schemas -Repeat the schema, table, sequence, function, and default-privilege grants for a custom queue-storage schema. Prepare custom schemas as the migrator; the runtime needs DML, sequence access, function execution, and `TRUNCATE`, but never DDL. See [Queue storage](../queue-storage-substrate.md). +Repeat the schema, table, sequence, function, and default-privilege grants for a custom queue-storage schema. Prepare custom schemas as the migrator; the runtime needs DML, sequence access, function execution, and `TRUNCATE`, but never DDL. Check lanes in a custom schema with `awa storage lane-sequences --schema `. See [Queue storage](../queue-storage-substrate.md). ## Why the grants are broad diff --git a/docs/upgrade-0.6-to-0.7.md b/docs/upgrade-0.6-to-0.7.md index 0afc48afb..7fff6173f 100644 --- a/docs/upgrade-0.6-to-0.7.md +++ b/docs/upgrade-0.6-to-0.7.md @@ -14,14 +14,41 @@ A [migrate-first rehearsal with released `awa-pg==0.6.7`](https://github.com/har verified a restricted worker completing jobs on the upgraded schema. That rehearsal does not cover concurrent mixed-version rollouts. -For a runtime without DDL privileges, apply v049 as the migrator, run -`awa storage prepare-queue --queue ` for every physical queue, and grant -sequence `USAGE, SELECT, UPDATE` before starting producers or workers. Match -shard and stripe counts and provision extra lanes before increasing them. External migration runners must apply the complete generated `awa migrate --sql --from 48 --to 49` bundle: it refreshes the installer -definition before replacing lane helpers in the installed schemas. -See [Database roles](security/database-roles.md#provision-queues-before-starting-restricted-runtimes). +definition before replacing lane helpers in the installed schemas. From v053, +runtimes without DDL privileges create lanes themselves through the schema +owner's provisioning functions, so `awa storage prepare-queue` is optional. + +## Lane ownership and v053 + +Before v053 a lane's sequences belonged to the role that first enqueued into, +claimed from or promoted into it. With several runtime roles, only roles whose +default privileges covered that role could use the lane; any other role failed +with `permission denied for sequence`. v053 installs `SECURITY DEFINER` +`provision_lane` and `provision_lane_sequences` functions, owned by the schema +owner, in each queue-storage schema. The lane helpers create sequences through +them, with grants that mirror the lane's head table. v053 then gives every +existing lane sequence to the schema owner and adds its missing grants, as far +as the migrating role is allowed to. + +0.6 and earlier 0.7 binaries call the same helpers, so lanes they create after +v053 are owned by the schema owner too. A 0.7 binary running ahead of v053 +detects that `provision_lane` is missing and creates lanes as before. + +To adopt it: + +1. Check the current state: `awa storage lane-sequences` (read-only, exits 2 + while lanes need repair). +2. Run `awa migrate` with the schema owner as the effective role, as described + in [Database roles](security/database-roles.md). +3. Read the migrate output. Any lane sequence it could not reassign is listed; + `awa storage lane-sequences` gives the statements to run and the role to run + them as. See + [Check and repair lane ownership](security/database-roles.md#check-and-repair-lane-ownership). + +External migration runners apply the generated `awa migrate --sql --from 52 +--to 53` bundle, then run `awa storage lane-sequences` to check for leftovers. ## Custom-schema discovery during migration diff --git a/skills/awa-operations/SKILL.md b/skills/awa-operations/SKILL.md index 4cfb039b0..5124fbc3f 100644 --- a/skills/awa-operations/SKILL.md +++ b/skills/awa-operations/SKILL.md @@ -91,11 +91,13 @@ Do not skip a major version. 0.5 → 0.7 is unsupported; step through 0.6. ## Restricted Database Roles -When deploying a runtime without schema CREATE, apply v049 and provision every -queue with `awa storage prepare-queue` as the migrator before starting producers -or workers. Match shard/stripe counts; provision additional lanes before changing -routing. Use the sequence grants and owner defaults in -[Database roles](../../docs/security/database-roles.md#provision-queues-before-starting-restricted-runtimes). +From v053 any runtime role creates lanes through the schema owner's +`SECURITY DEFINER` provisioning functions, so runtimes need neither schema +CREATE nor sequence default privileges, and `awa storage prepare-queue` is +optional. Run migrations with the schema owner as the effective role. Check +lane ownership with `awa storage lane-sequences` (exits 2 while lanes need +repair) and fix it with `--repair` or the statements it prints; see +[Database roles](../../docs/security/database-roles.md#check-and-repair-lane-ownership). ## Rolling Upgrades From 70fe5ff66ae289189b3b3340737280f2d316372e Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Fri, 6 Nov 2026 13:35:46 +1300 Subject: [PATCH 2/5] docs(migrations): link v053 to its pull request --- awa-model/migrations/v053_definer_lane_provisioning.sql | 1 + 1 file changed, 1 insertion(+) diff --git a/awa-model/migrations/v053_definer_lane_provisioning.sql b/awa-model/migrations/v053_definer_lane_provisioning.sql index a56d96978..758627703 100644 --- a/awa-model/migrations/v053_definer_lane_provisioning.sql +++ b/awa-model/migrations/v053_definer_lane_provisioning.sql @@ -1,4 +1,5 @@ -- v053: Create queue lanes as the schema owner, whichever role touches them first. +-- Tracking: https://github.com/hardbyte/awa/pull/637 -- Lane sequences used to belong to the role that first enqueued into, claimed -- from or promoted into a lane, and only that role (plus whoever its default -- privileges named) could use them. refresh_lane_sequence_helpers now installs From 80178b39b6b2c40af8430809667a0a8a2b701069 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Fri, 6 Nov 2026 14:04:15 +1300 Subject: [PATCH 3/5] fix(queue-storage): accept every lane the invoker path accepts in provision_lane Queue storage takes any queue name, including an empty one, and any smallint priority, so the provisioning functions only reject a null argument or a shard outside 0-63. --- awa-model/migrations/v023_install_queue_storage_substrate.sql | 4 ++-- docs/security/database-roles.md | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/awa-model/migrations/v023_install_queue_storage_substrate.sql b/awa-model/migrations/v023_install_queue_storage_substrate.sql index b5526cccc..39df93e86 100644 --- a/awa-model/migrations/v023_install_queue_storage_substrate.sql +++ b/awa-model/migrations/v023_install_queue_storage_substrate.sql @@ -272,8 +272,8 @@ BEGIN v_seq_name TEXT; v_grant TEXT; BEGIN - IF p_queue IS NULL OR p_queue = '' - OR p_priority IS NULL OR p_priority NOT BETWEEN 1 AND 4 + IF p_queue IS NULL + OR p_priority IS NULL OR p_enqueue_shard IS NULL OR p_enqueue_shard NOT BETWEEN 0 AND 63 THEN RAISE EXCEPTION 'invalid queue lane: queue %%, priority %%, enqueue shard %%', quote_nullable(p_queue), p_priority, p_enqueue_shard diff --git a/docs/security/database-roles.md b/docs/security/database-roles.md index 1b5e0091f..5f15793e7 100644 --- a/docs/security/database-roles.md +++ b/docs/security/database-roles.md @@ -65,7 +65,7 @@ If an existing installation created objects as `awa_migrator` without first sett ## Lane sequences -Each queue lane, `(queue, priority, enqueue_shard)`, has an enqueue and a claim sequence. From schema v053 they are created by the schema owner, whichever role first enqueues into, claims from or promotes into the lane: the lane helpers call `provision_lane` and `provision_lane_sequences` in the queue-storage schema. These are `SECURITY DEFINER` functions owned by the schema owner with a pinned `search_path`. Like the other runtime-callable functions they are executable by `PUBLIC`, so schema `USAGE` is what gates them, and they only create lanes, after validating the queue, priority (1–4) and shard (0–63). A runtime therefore needs neither schema `CREATE` nor default privileges to start using a new queue, priority or shard. +Each queue lane, `(queue, priority, enqueue_shard)`, has an enqueue and a claim sequence. From schema v053 they are created by the schema owner, whichever role first enqueues into, claims from or promotes into the lane: the lane helpers call `provision_lane` and `provision_lane_sequences` in the queue-storage schema. These are `SECURITY DEFINER` functions owned by the schema owner with a pinned `search_path`. Like the other runtime-callable functions they are executable by `PUBLIC`, so schema `USAGE` is what gates them, and they only create lanes, after checking that the queue, priority and shard are set and the shard is 0–63. A runtime therefore needs neither schema `CREATE` nor default privileges to start using a new queue, priority or shard. A new lane sequence gets the privileges of the head table it backs. A role that may `INSERT` or `UPDATE` `queue_enqueue_heads` gets `USAGE, SELECT, UPDATE` on the enqueue sequences, and one that may only `SELECT` it gets `SELECT`; `queue_claim_heads` does the same for the claim sequences. Granting the runtime the table privileges above is enough, and no sequence default privileges are needed for lanes. From 9a7eed99ae2fa439b9488e728b9129dac9c4917f Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Fri, 6 Nov 2026 18:21:00 +1300 Subject: [PATCH 4/5] fix(queue-storage): repair lane sequences per transaction with a bounded lock wait v053 reassigned lane sequences inside the migration transaction. ALTER SEQUENCE ... OWNER takes the sequence's ACCESS EXCLUSIVE lock until commit, with no lock_timeout, so an open enqueue on one lane could stall the whole migration and every repaired lane waited for the upgrade to commit. v053 now only refreshes the helpers. After it commits, awa migrate repairs each lane sequence with its own statement through awa.repair_lane_sequence(schema, sequence), which has lock_timeout = 300ms; on 55P03 the sequence is left unchanged and reported as busy. `awa storage lane-sequences --repair` uses the same path, and the set-returning awa.repair_lane_sequences(schema) is gone. Reassigning needs the current owner's privileges, which the schema owner lacks for sequences the migration role or an app role created. A migration role connected with role set to the owner therefore could not reassign its own sequences. The repair now retries such a sequence as the login role, lane_sequence_problems reports repair_roles (the login roles that can repair each sequence), and leftovers are reported with the next step to take. The docs state which role to run the repair as. --- CHANGELOG.md | 13 +- awa-cli/src/main.rs | 61 +++-- .../v023_install_queue_storage_substrate.sql | 122 +++++---- .../v053_definer_lane_provisioning.sql | 30 +-- awa-model/src/migrations.rs | 62 ++++- awa-model/src/storage.rs | 237 ++++++++++++++++-- awa/tests/lane_ownership_roles_test.rs | 194 ++++++++++---- docs/security/database-roles.md | 27 +- docs/upgrade-0.6-to-0.7.md | 17 +- skills/awa-operations/SKILL.md | 3 +- 10 files changed, 577 insertions(+), 189 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6780d6113..7f8ce530a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,14 +18,17 @@ Notable changes between releases. Detailed migration notes for storage transitio the privileges of the head table it backs, explicitly rather than through default privileges. A runtime without schema `CREATE` can therefore create lanes itself, and `awa storage prepare-queue` is optional and may run as any - runtime role. v053 gives existing lane sequences to the schema owner and adds - their missing grants, one sequence at a time and only where the migrating - role is allowed to. It never fails on a sequence it cannot reassign. The new + runtime role. After v053 commits, `awa migrate` gives existing lane + sequences to the schema owner and adds their missing grants, one sequence per + transaction with a 300 ms lock wait, and only where it is allowed to, trying + the effective role and then the login role. A sequence an open transaction + holds is skipped and reported as busy; one it may not change is reported + with the roles that can repair it. Neither fails the migration. The new read-only `awa storage lane-sequences` (or `awa.lane_sequence_problems(schema)`) lists what remains, with the statements to run and the roles to run them as, and exits 2 while any remain; - `--repair` (or `awa.repair_lane_sequences(schema)`) repairs what the - connected role may. `awa migrate` prints a summary of the leftovers. A binary + `--repair` (or `awa.repair_lane_sequence(schema, sequence)` per sequence) + repairs what the connected role may. `awa migrate` prints a summary of the leftovers. A binary running ahead of v053 detects that `provision_lane` is missing and creates lanes as before, and older binaries create correctly owned lanes once v053 has refreshed the helpers they call. See "Lane ownership and v053" in the diff --git a/awa-cli/src/main.rs b/awa-cli/src/main.rs index 5e53264c5..e111c3574 100644 --- a/awa-cli/src/main.rs +++ b/awa-cli/src/main.rs @@ -273,6 +273,23 @@ fn sanitize_filename(description: &str) -> String { /// mutating commands require an explicit `--context` or `--database-url` /// once more than one context is defined, and are confirmation-gated on /// contexts marked `production = true`. +/// After a migration, say which lane sequences still need repair and how to +/// finish it. Read-only. +async fn print_lane_sequence_leftovers( + pool: &sqlx::PgPool, +) -> Result<(), Box> { + let mut problems = Vec::new(); + for (schema, _) in awa_model::storage::lane_sequence_problem_counts(pool).await? { + problems.extend(awa_model::storage::lane_sequence_problems(pool, &schema).await?); + } + for next_step in awa_model::storage::lane_sequence_next_steps( + problems.iter().map(|problem| (problem, false)), + ) { + eprintln!("warning: {next_step}"); + } + Ok(()) +} + fn access_for(command: &Commands) -> context::Access { use context::Access::{Mutating, ReadOnly}; match command { @@ -834,7 +851,8 @@ enum StorageCommands { /// Queue-storage schema to check #[arg(long, default_value = "awa")] schema: String, - /// Reassign and grant what the connected role is allowed to first + /// Reassign and grant first, as far as the connected role (or, failing + /// that, its login role) is allowed to #[arg(long)] repair: bool, }, @@ -1194,16 +1212,7 @@ async fn run() -> Result<(), Box> { }; awa_model::migrations::run_with_options(&pool, options).await?; println!("Migrations applied successfully."); - for (schema, count) in - awa_model::storage::lane_sequence_problem_counts(&pool).await? - { - eprintln!( - "warning: {count} lane sequence(s) in schema {schema} are not owned by \ - the schema owner or lack runtime grants, and the migrating role could \ - not repair them. Run `awa storage lane-sequences --schema {schema}` for \ - the statements to run and the roles to run them as." - ); - } + print_lane_sequence_leftovers(&pool).await?; } } @@ -1956,19 +1965,18 @@ async fn run() -> Result<(), Box> { ); } StorageCommands::LaneSequences { schema, repair } => { - let remaining = if repair { - let results = - awa_model::storage::repair_lane_sequences(&pool, &schema).await?; + let next_steps = if repair { + let mut conn = pool.acquire().await?; + let repairs = + awa_model::storage::repair_lane_sequences(&mut conn, &schema) + .await?; println!( "{}", serde_json::to_string_pretty(&serde_json::json!({ - "schema": schema, "results": results, + "schema": schema, "results": repairs, }))? ); - results - .iter() - .filter(|result| result.outcome != "repaired") - .count() + awa_model::storage::lane_sequence_repair_next_steps(&repairs) } else { let problems = awa_model::storage::lane_sequence_problems(&pool, &schema).await?; @@ -1978,15 +1986,14 @@ async fn run() -> Result<(), Box> { "schema": schema, "problems": problems, }))? ); - problems.len() + awa_model::storage::lane_sequence_next_steps( + problems.iter().map(|problem| (problem, false)), + ) }; - if remaining > 0 { - eprintln!( - "{remaining} lane sequence(s) in schema {schema} need repair. Run \ - each missing_grants statement as the sequence owner (or a role \ - that can SET ROLE to it), and each reassign_sql statement as a \ - role that can SET ROLE to both the owner and expected_owner." - ); + if !next_steps.is_empty() { + for next_step in next_steps { + eprintln!("{next_step}"); + } std::process::exit(2); } } diff --git a/awa-model/migrations/v023_install_queue_storage_substrate.sql b/awa-model/migrations/v023_install_queue_storage_substrate.sql index 39df93e86..f3736cef1 100644 --- a/awa-model/migrations/v023_install_queue_storage_substrate.sql +++ b/awa-model/migrations/v023_install_queue_storage_substrate.sql @@ -407,15 +407,18 @@ REVOKE EXECUTE ON FUNCTION awa.refresh_lane_sequence_helpers(TEXT) FROM PUBLIC; -- USAGE, SELECT and UPDATE on it, and whoever may read it gets SELECT: -- queue_enqueue_seq_* mirror queue_enqueue_heads, queue_claim_seq_* mirror -- queue_claim_heads. missing_grants holds the statements to run as the --- sequence owner; reassign_sql, when the owner is wrong, needs a role that can --- SET ROLE to both the current owner and the schema owner. Read-only. +-- sequence owner; reassign_sql, when the owner is wrong, needs a role that +-- holds the current owner's privileges and can SET ROLE to the schema owner. +-- repair_roles lists the login roles, superusers aside, that can run both. +-- Read-only. CREATE OR REPLACE FUNCTION awa.lane_sequence_problems(p_schema TEXT, p_sequence TEXT DEFAULT NULL) RETURNS TABLE ( sequence_name TEXT, owner TEXT, expected_owner TEXT, missing_grants TEXT[], - reassign_sql TEXT + reassign_sql TEXT, + repair_roles TEXT[] ) LANGUAGE sql STABLE @@ -489,55 +492,76 @@ AS $problems$ CASE WHEN lane_sequence.relowner <> lane_sequence.nspowner THEN format( 'ALTER SEQUENCE %I.%I OWNER TO %I', p_schema, lane_sequence.relname, pg_get_userbyid(lane_sequence.nspowner) - ) END + ) END, + ARRAY( + SELECT candidate.rolname::text + FROM pg_roles AS candidate + WHERE candidate.rolcanlogin + AND NOT candidate.rolsuper + AND pg_has_role(candidate.oid, lane_sequence.relowner, 'USAGE') + AND (lane_sequence.relowner = lane_sequence.nspowner + OR (pg_has_role(candidate.oid, lane_sequence.nspowner, 'USAGE') + AND pg_has_role( + candidate.oid, + lane_sequence.nspowner, + CASE WHEN current_setting('server_version_num')::int >= 160000 + THEN 'SET' ELSE 'MEMBER' END + ))) + ORDER BY candidate.rolname + ) FROM lane_sequence WHERE lane_sequence.relowner <> lane_sequence.nspowner OR EXISTS (SELECT 1 FROM missing WHERE missing.oid = lane_sequence.oid) ORDER BY lane_sequence.relname $problems$; --- Give every lane sequence in p_schema to the schema owner and add its missing --- runtime grants, one sequence at a time, doing only what the calling role is --- allowed to: a sequence it cannot reassign keeps its owner and still gets --- the grants when the caller can act as that owner. Never raises for a --- sequence it cannot repair. Returns one row per sequence that had a problem: --- outcome is 'repaired', 'owner_unchanged' (grants complete, owner still --- wrong) or 'unrepaired', with the statements still to run. -CREATE OR REPLACE FUNCTION awa.repair_lane_sequences(p_schema TEXT) +-- Give one lane sequence to the schema owner and add its missing runtime +-- grants, doing only what the calling role is allowed to: a sequence it +-- cannot reassign keeps its owner and still gets the grants when the caller +-- holds that owner's privileges. Never raises for a sequence it cannot +-- repair. Reassigning takes the sequence's ACCESS EXCLUSIVE lock, which an +-- open enqueue or claim on the lane conflicts with, so the lock wait is +-- bounded: on timeout nothing changes and the outcome is 'busy'. The lock is +-- held until the caller's transaction ends, so callers repair one sequence +-- per transaction. Returns no row when the sequence needs nothing, otherwise +-- one row: outcome is 'repaired', 'owner_unchanged' (grants complete, owner +-- still wrong), 'unrepaired' or 'busy', with the statements still to run. +CREATE OR REPLACE FUNCTION awa.repair_lane_sequence(p_schema TEXT, p_sequence TEXT) RETURNS TABLE ( sequence_name TEXT, owner TEXT, expected_owner TEXT, outcome TEXT, missing_grants TEXT[], - reassign_sql TEXT + reassign_sql TEXT, + repair_roles TEXT[] ) LANGUAGE plpgsql SECURITY INVOKER SET search_path = pg_catalog, pg_temp +SET lock_timeout = '300ms' AS $repair$ DECLARE v_problem RECORD; v_remaining RECORD; - v_has_remaining BOOLEAN; - v_owner OID; + v_busy BOOLEAN := false; v_set_mode TEXT := CASE WHEN current_setting('server_version_num')::int >= 160000 THEN 'SET' ELSE 'MEMBER' END; v_statement TEXT; BEGIN - FOR v_problem IN - SELECT * FROM awa.lane_sequence_problems(p_schema) AS problem - LOOP - SELECT sequence.relowner INTO v_owner - FROM pg_class AS sequence - WHERE sequence.oid = to_regclass(format('%I.%I', p_schema, v_problem.sequence_name)); + SELECT * INTO v_problem + FROM awa.lane_sequence_problems(p_schema, p_sequence) AS problem; + IF NOT FOUND THEN + RETURN; + END IF; + BEGIN -- ALTER ... OWNER TO needs the current owner's privileges, SET ROLE to -- the new owner, and the new owner's CREATE on the schema (which a -- schema owner has). Check first so an unrelated owner is skipped - -- rather than raising. + -- without waiting for the lock. IF v_problem.reassign_sql IS NOT NULL - AND pg_has_role(current_user, v_owner, 'USAGE') + AND pg_has_role(current_user, to_regrole(quote_ident(v_problem.owner)), 'USAGE') AND pg_has_role(current_user, to_regrole(quote_ident(v_problem.expected_owner)), v_set_mode) THEN BEGIN EXECUTE v_problem.reassign_sql; @@ -547,10 +571,8 @@ BEGIN END IF; SELECT * INTO v_remaining - FROM awa.lane_sequence_problems(p_schema, v_problem.sequence_name) AS problem; - v_has_remaining := FOUND; - - IF v_has_remaining + FROM awa.lane_sequence_problems(p_schema, p_sequence) AS problem; + IF FOUND AND cardinality(v_remaining.missing_grants) > 0 AND pg_has_role(current_user, to_regrole(quote_ident(v_remaining.owner)), 'USAGE') THEN FOREACH v_statement IN ARRAY v_remaining.missing_grants LOOP @@ -560,31 +582,35 @@ BEGIN NULL; END; END LOOP; - SELECT * INTO v_remaining - FROM awa.lane_sequence_problems(p_schema, v_problem.sequence_name) AS problem; - v_has_remaining := FOUND; END IF; - - sequence_name := v_problem.sequence_name; - expected_owner := v_problem.expected_owner; - IF v_has_remaining THEN - owner := v_remaining.owner; - missing_grants := v_remaining.missing_grants; - reassign_sql := v_remaining.reassign_sql; - outcome := CASE WHEN cardinality(v_remaining.missing_grants) = 0 - THEN 'owner_unchanged' ELSE 'unrepaired' END; - ELSE - owner := v_problem.expected_owner; - missing_grants := ARRAY[]::TEXT[]; - reassign_sql := NULL; - outcome := 'repaired'; - END IF; - RETURN NEXT; - END LOOP; + EXCEPTION WHEN lock_not_available THEN + v_busy := true; + END; + + SELECT * INTO v_remaining + FROM awa.lane_sequence_problems(p_schema, p_sequence) AS problem; + sequence_name := v_problem.sequence_name; + expected_owner := v_problem.expected_owner; + IF FOUND THEN + owner := v_remaining.owner; + missing_grants := v_remaining.missing_grants; + reassign_sql := v_remaining.reassign_sql; + repair_roles := v_remaining.repair_roles; + outcome := CASE WHEN v_busy THEN 'busy' + WHEN cardinality(v_remaining.missing_grants) = 0 THEN 'owner_unchanged' + ELSE 'unrepaired' END; + ELSE + owner := v_problem.expected_owner; + missing_grants := ARRAY[]::TEXT[]; + reassign_sql := NULL; + repair_roles := ARRAY[]::TEXT[]; + outcome := 'repaired'; + END IF; + RETURN NEXT; END $repair$; -REVOKE EXECUTE ON FUNCTION awa.repair_lane_sequences(TEXT) FROM PUBLIC; +REVOKE EXECUTE ON FUNCTION awa.repair_lane_sequence(TEXT, TEXT) FROM PUBLIC; -- Retained job output (ADR-048): add the output columns to an existing -- substrate and expose them through terminal_jobs. Called by the installer diff --git a/awa-model/migrations/v053_definer_lane_provisioning.sql b/awa-model/migrations/v053_definer_lane_provisioning.sql index 758627703..2eba66ff6 100644 --- a/awa-model/migrations/v053_definer_lane_provisioning.sql +++ b/awa-model/migrations/v053_definer_lane_provisioning.sql @@ -6,26 +6,24 @@ -- two SECURITY DEFINER functions, provision_lane and provision_lane_sequences, -- owned by the schema owner, and the lane helpers create a missing sequence -- through them. Each new sequence gets the grants of the head table it backs. --- This migration then repairs existing lane sequences: it gives each one to --- the schema owner and adds its missing grants, one sequence at a time and --- only where the migrating role may. Sequences it cannot reassign are reported --- in a warning and by awa.lane_sequence_problems(); the migration still commits. +-- This migration does not reassign existing lane sequences: ALTER SEQUENCE +-- OWNER takes the sequence's ACCESS EXCLUSIVE lock until commit, and the +-- migration runs in one transaction. After it commits, `awa migrate` repairs +-- each lane sequence in its own short transaction with a bounded lock wait +-- (awa.repair_lane_sequence). Another runner can do the same with +-- `awa storage lane-sequences --repair`. -- -- N-1 binaries call the same helpers with the same arguments and cursor -- semantics, so lanes they create are owned by the schema owner too. A binary -- that predates provision_lane keeps using the helpers; this binary probes for -- provision_lane and uses the helpers when it is missing. -- --- Live load: function replacement, plus ALTER SEQUENCE OWNER on the lane --- sequences that need it. That takes each such sequence's ACCESS EXCLUSIVE --- lock until commit, so enqueues and claims on those lanes wait for the --- migration to commit; correctly owned lanes are not touched. Reads the --- catalog only; no table DDL or row scans. +-- Live load: function replacement only. Reads the catalog; no table DDL, row +-- scans or sequence locks. DO $$ DECLARE v_schema TEXT; - v_leftovers TEXT; v_count BIGINT; BEGIN FOR v_schema IN @@ -41,17 +39,13 @@ BEGIN PERFORM awa.refresh_lane_sequence_helpers(v_schema); - SELECT count(*), - string_agg(format('%s (owner %s, %s)', repair.sequence_name, repair.owner, repair.outcome), - ', ' ORDER BY repair.sequence_name) - INTO v_count, v_leftovers - FROM awa.repair_lane_sequences(v_schema) AS repair - WHERE repair.outcome <> 'repaired'; + SELECT count(*) INTO v_count + FROM awa.lane_sequence_problems(v_schema) AS problem; IF v_count > 0 THEN - RAISE WARNING '% lane sequence(s) in schema % still need repair: %', v_count, v_schema, v_leftovers + RAISE NOTICE '% lane sequence(s) in schema % are not owned by the schema owner or lack runtime grants', v_count, v_schema USING HINT = format( - 'Run awa storage lane-sequences --schema %s for the statements to run and the role to run them as.', + 'awa migrate repairs them after this commits. Otherwise run awa storage lane-sequences --schema %s --repair as the migration role itself, without SET ROLE.', v_schema ); END IF; diff --git a/awa-model/src/migrations.rs b/awa-model/src/migrations.rs index 51c423222..f30033198 100644 --- a/awa-model/src/migrations.rs +++ b/awa-model/src/migrations.rs @@ -423,9 +423,9 @@ pub async fn run_with_options(pool: &PgPool, options: MigrateOptions) -> Result< // only sound because every migration is transaction-safe (no // `CREATE INDEX CONCURRENTLY`, `VACUUM`, or explicit transaction control). let mut retry = 0; - loop { + let version_before = loop { match run_migration_transaction(pool, &options, lock_key).await { - Ok(()) => break, + Ok(version_before) => break version_before, Err(error) if is_migration_deadlock(&error) && retry < MIGRATION_DEADLOCK_RETRIES => { retry += 1; let delay = std::time::Duration::from_millis(50 * (1 << (retry - 1))); @@ -439,6 +439,10 @@ pub async fn run_with_options(pool: &PgPool, options: MigrateOptions) -> Result< } Err(error) => return Err(error), } + }; + + if version_before < LANE_PROVISIONING_VERSION { + repair_lane_sequences_after_migration(pool).await; } // Best-effort admin-metadata cache warmup, deliberately outside the @@ -449,29 +453,73 @@ pub async fn run_with_options(pool: &PgPool, options: MigrateOptions) -> Result< Ok(()) } +/// Returns the schema version the transaction started from. async fn run_migration_transaction( pool: &PgPool, options: &MigrateOptions, lock_key: i64, -) -> Result<(), AwaError> { +) -> Result { let mut tx = pool.begin().await?; sqlx::query("SELECT pg_advisory_xact_lock($1)") .bind(lock_key) .execute(&mut *tx) .await?; - apply_migrations(&mut tx, options).await?; + let version_before = apply_migrations(&mut tx, options).await?; tx.commit().await?; - Ok(()) + Ok(version_before) +} + +/// The migration that makes lane sequences belong to the schema owner. +const LANE_PROVISIONING_VERSION: i32 = 53; + +/// Repair existing lane sequences once v053 has committed. +/// +/// `ALTER SEQUENCE ... OWNER` takes the sequence's ACCESS EXCLUSIVE lock until +/// its transaction ends, so it cannot run inside the migration transaction +/// without stalling every enqueue and claim on those lanes until the whole +/// upgrade commits. Each sequence is repaired in its own short transaction +/// with a bounded lock wait instead. Best-effort: the schema is committed, and +/// whatever is left is reported with the command that finishes it. +/// +/// Uses a detached connection for the reason given on +/// [`warm_admin_metadata_cache`]. +async fn repair_lane_sequences_after_migration(pool: &PgPool) { + let mut conn = match pool.acquire().await { + Ok(conn) => conn.detach(), + Err(error) => { + warn!(%error, "Could not connect to repair lane sequences; run `awa storage lane-sequences --repair`"); + return; + } + }; + match crate::storage::repair_lane_sequences_in_every_schema(&mut conn).await { + Ok(repairs) => { + let repaired = repairs + .iter() + .filter(|repair| repair.outcome == crate::storage::LaneSequenceOutcome::Repaired) + .count(); + if repaired > 0 { + info!(repaired, "Gave lane sequences to the schema owner"); + } + for next_step in crate::storage::lane_sequence_repair_next_steps(&repairs) { + warn!("{next_step}"); + } + } + Err(error) => { + warn!(%error, "Could not repair lane sequences; run `awa storage lane-sequences --repair`"); + } + } + let _ = conn.close().await; } fn is_migration_deadlock(error: &AwaError) -> bool { error.is_deadlock() } +/// Returns the schema version before this run. async fn apply_migrations( conn: &mut PgConnection, options: &MigrateOptions, -) -> Result<(), AwaError> { +) -> Result { let has_schema: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM pg_namespace WHERE nspname = 'awa')") .fetch_one(&mut *conn) @@ -564,7 +612,7 @@ async fn apply_migrations( info!(version = current, "Schema is up to date"); } - Ok(()) + Ok(current) } /// Warm the admin-metadata cache after migrations commit. diff --git a/awa-model/src/storage.rs b/awa-model/src/storage.rs index d67dc2048..14770c40a 100644 --- a/awa-model/src/storage.rs +++ b/awa-model/src/storage.rs @@ -2,7 +2,7 @@ use crate::admin; use crate::error::AwaError; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; -use sqlx::{FromRow, PgExecutor, PgPool}; +use sqlx::{Connection, FromRow, PgConnection, PgExecutor, PgPool}; use std::collections::BTreeMap; #[derive(Debug, Clone, Serialize, Deserialize, FromRow, PartialEq)] @@ -472,30 +472,65 @@ pub async fn status_report(pool: &PgPool) -> Result, pub reassign_sql: Option, + /// Login roles, superusers aside, that can run every statement above. + /// Empty when no such role exists yet. + pub repair_roles: Vec, } -/// What [`repair_lane_sequences`] did to one lane sequence that had a problem. +/// What a repair did to one lane sequence. +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Hash)] +#[serde(rename_all = "snake_case")] +pub enum LaneSequenceOutcome { + /// Owned by the schema owner with every runtime grant. + Repaired, + /// Grants complete, but the role could not reassign the sequence. + OwnerUnchanged, + /// The role could neither reassign the sequence nor add its grants. + Unrepaired, + /// An open transaction held the sequence's lock past the repair's lock + /// timeout, so nothing changed. Rerun the repair. + Busy, +} + +impl TryFrom for LaneSequenceOutcome { + type Error = String; + + fn try_from(outcome: String) -> Result { + match outcome.as_str() { + "repaired" => Ok(Self::Repaired), + "owner_unchanged" => Ok(Self::OwnerUnchanged), + "unrepaired" => Ok(Self::Unrepaired), + "busy" => Ok(Self::Busy), + _ => Err(format!("unknown lane sequence repair outcome {outcome:?}")), + } + } +} + +/// What [`repair_lane_sequences`] did to one lane sequence that had a problem, +/// with what is still wrong afterwards. #[derive(Debug, Clone, Serialize, Deserialize, FromRow, PartialEq, Eq)] pub struct LaneSequenceRepair { + pub schema: String, pub sequence_name: String, /// Owner after the repair. pub owner: String, pub expected_owner: String, - /// `repaired`; `owner_unchanged` when the grants are complete but the - /// connected role could not reassign the sequence; or `unrepaired` when - /// grants are still missing too. - pub outcome: String, + #[sqlx(try_from = "String")] + pub outcome: LaneSequenceOutcome, pub missing_grants: Vec, pub reassign_sql: Option, + /// Login roles, superusers aside, that can finish the repair. + pub repair_roles: Vec, } /// Lane sequences in `schema` that need repair. Read-only; any role may run it. @@ -507,7 +542,8 @@ where E: PgExecutor<'e>, { Ok(sqlx::query_as( - "SELECT sequence_name, owner, expected_owner, missing_grants, reassign_sql \ + "SELECT $1::text AS schema, sequence_name, owner, expected_owner, missing_grants, \ + reassign_sql, repair_roles \ FROM awa.lane_sequence_problems($1)", ) .bind(schema) @@ -516,25 +552,182 @@ where } /// Give each lane sequence in `schema` to the schema owner and add its missing -/// grants, as far as the connected role is allowed to. A sequence the role -/// cannot reassign is returned with the statements still to run, never an -/// error. +/// grants, as far as the connection's roles are allowed to. A sequence that +/// cannot be repaired is returned with the statements still to run and the +/// roles that can run them, never an error. +/// +/// Each sequence is repaired by its own statement, outside any transaction +/// the caller holds, so a reassignment's ACCESS EXCLUSIVE lock is released +/// before the next sequence and waits at most the repair function's lock +/// timeout. A sequence locked for longer by an open enqueue or claim comes +/// back [`LaneSequenceOutcome::Busy`]. +/// +/// Reassigning needs the current owner's privileges. A migration role that +/// connects as itself and sets its role to the schema owner (`role` in the +/// connection options) loses its own privileges while acting as the owner, +/// so a sequence the current role cannot repair is retried once as the login +/// role. pub async fn repair_lane_sequences( - pool: &PgPool, + conn: &mut PgConnection, schema: &str, ) -> Result, AwaError> { - Ok(sqlx::query_as( - "SELECT sequence_name, owner, expected_owner, outcome, missing_grants, reassign_sql \ - FROM awa.repair_lane_sequences($1)", - ) - .bind(schema) - .fetch_all(pool) - .await?) + let (current_role, login_role): (String, String) = + sqlx::query_as("SELECT current_user::text, session_user::text") + .fetch_one(&mut *conn) + .await?; + let problems = lane_sequence_problems(&mut *conn, schema).await?; + let mut repairs = Vec::with_capacity(problems.len()); + for problem in problems { + let mut repair: Option = sqlx::query_as(REPAIR_LANE_SEQUENCE_SQL) + .bind(schema) + .bind(&problem.sequence_name) + .fetch_optional(&mut *conn) + .await?; + let role_was_the_obstacle = matches!( + repair.as_ref().map(|repair| repair.outcome), + Some(LaneSequenceOutcome::OwnerUnchanged | LaneSequenceOutcome::Unrepaired) + ); + if role_was_the_obstacle && login_role != current_role { + if let Some(retried) = + repair_lane_sequence_as_login_role(conn, schema, &problem.sequence_name).await? + { + repair = retried; + } + } + repairs.extend(repair); + } + Ok(repairs) +} + +const REPAIR_LANE_SEQUENCE_SQL: &str = + "SELECT $1::text AS schema, sequence_name, owner, expected_owner, outcome, missing_grants, \ + reassign_sql, repair_roles \ + FROM awa.repair_lane_sequence($1, $2)"; + +/// Repair one sequence with the login role as the current role, for this +/// transaction only. `None` when the login role may not run the repair. +async fn repair_lane_sequence_as_login_role( + conn: &mut PgConnection, + schema: &str, + sequence_name: &str, +) -> Result>, AwaError> { + let mut transaction = conn.begin().await?; + sqlx::query("SET LOCAL ROLE NONE") + .execute(&mut *transaction) + .await?; + let repair = sqlx::query_as(REPAIR_LANE_SEQUENCE_SQL) + .bind(schema) + .bind(sequence_name) + .fetch_optional(&mut *transaction) + .await; + match repair { + Ok(repair) => { + transaction.commit().await?; + Ok(Some(repair)) + } + Err(sqlx::Error::Database(error)) if error.code().as_deref() == Some("42501") => { + transaction.rollback().await?; + Ok(None) + } + Err(error) => Err(error.into()), + } +} + +/// [`repair_lane_sequences`] for every queue-storage schema with a lane +/// sequence that needs repair. +pub async fn repair_lane_sequences_in_every_schema( + conn: &mut PgConnection, +) -> Result, AwaError> { + let mut repairs = Vec::new(); + for (schema, _) in lane_sequence_problem_counts(&mut *conn).await? { + repairs.extend(repair_lane_sequences(conn, &schema).await?); + } + Ok(repairs) +} + +/// One line per group of lane sequences that still need repair, saying what +/// to do next. `busy` marks sequences a repair found locked. +pub fn lane_sequence_next_steps<'a>( + leftovers: impl IntoIterator, +) -> Vec { + let mut groups: BTreeMap<(String, bool, String, String, Vec), usize> = BTreeMap::new(); + for (problem, busy) in leftovers { + *groups + .entry(( + problem.schema.clone(), + busy, + problem.owner.clone(), + problem.expected_owner.clone(), + problem.repair_roles.clone(), + )) + .or_default() += 1; + } + groups + .into_iter() + .map( + |((schema, busy, owner, expected_owner, repair_roles), count)| { + let rerun = format!("awa storage lane-sequences --schema {schema} --repair"); + if busy { + format!( + "{count} lane sequence(s) in schema {schema} owned by {owner} are busy: an \ + open transaction held the lock, so nothing changed. Rerun `{rerun}`." + ) + } else if repair_roles.is_empty() { + format!( + "{count} lane sequence(s) in schema {schema} owned by {owner} still need \ + repair, and no login role can both act as {owner} and SET ROLE to \ + {expected_owner}. Grant {owner} to the migration role, then rerun `{rerun}` \ + connected as the migration role, or run the statements from \ + `awa storage lane-sequences --schema {schema}` as {owner}." + ) + } else { + format!( + "{count} lane sequence(s) in schema {schema} owned by {owner} still need \ + repair. Rerun `{rerun}` connected as one of: {}.", + repair_roles.join(", ") + ) + } + }, + ) + .collect() +} + +impl LaneSequenceRepair { + /// The problem left after this repair, or `None` when it was repaired. + pub fn remaining_problem(&self) -> Option { + (self.outcome != LaneSequenceOutcome::Repaired).then(|| LaneSequenceProblem { + schema: self.schema.clone(), + sequence_name: self.sequence_name.clone(), + owner: self.owner.clone(), + expected_owner: self.expected_owner.clone(), + missing_grants: self.missing_grants.clone(), + reassign_sql: self.reassign_sql.clone(), + repair_roles: self.repair_roles.clone(), + }) + } +} + +/// [`lane_sequence_next_steps`] for the sequences a repair left behind. +pub fn lane_sequence_repair_next_steps(repairs: &[LaneSequenceRepair]) -> Vec { + let leftovers: Vec<(LaneSequenceProblem, bool)> = repairs + .iter() + .filter_map(|repair| { + repair + .remaining_problem() + .map(|problem| (problem, repair.outcome == LaneSequenceOutcome::Busy)) + }) + .collect(); + lane_sequence_next_steps(leftovers.iter().map(|(problem, busy)| (problem, *busy))) } /// How many lane sequences need repair in each queue-storage schema that has /// any. Read-only. -pub async fn lane_sequence_problem_counts(pool: &PgPool) -> Result, AwaError> { +pub async fn lane_sequence_problem_counts<'e, E>( + executor: E, +) -> Result, AwaError> +where + E: PgExecutor<'e>, +{ Ok(sqlx::query_as( "SELECT discovered.schema_name, count(*) \ FROM awa.queue_storage_schemas() AS discovered \ @@ -542,6 +735,6 @@ pub async fn lane_sequence_problem_counts(pool: &PgPool) -> Result Result { + PgPoolOptions::new() + .max_connections(1) + .connect_with( + self.connect_options + .clone() + .username(&self.migrator_name) + .options([("role", self.owner_name.as_str())]), + ) + .await + } + + /// The `queue_enqueue_seq_*` sequence of `queue`'s first lane. + async fn enqueue_sequence(&self, queue: &str) -> Result { + Ok(self + .lane_sequence_owners(queue) + .await? + .into_iter() + .map(|(sequence, _)| sequence) + .find(|sequence| sequence.starts_with("queue_enqueue_seq_")) + .expect("lane has an enqueue sequence")) + } + /// `(sequence, owner)` for every lane sequence of `queue`. async fn lane_sequence_owners( &self, @@ -590,20 +618,15 @@ async fn migration_repairs_lanes_it_may_and_reports_the_rest() -> TestResult { )))) .await?; } - let failed = drain(&cluster.app_b, "owner_lane").await.unwrap_err(); - assert!( - failed - .to_string() - .contains("permission denied for sequence"), - "{failed}" - ); - let failed = drain(&cluster.app_b, "app_lane").await.unwrap_err(); - assert!( - failed - .to_string() - .contains("permission denied for sequence"), - "{failed}" - ); + for lane in ["owner_lane", "app_lane"] { + let failed = drain(&cluster.app_b, lane).await.unwrap_err(); + assert!( + failed + .to_string() + .contains("permission denied for sequence"), + "{failed}" + ); + } // 0.7 refuses to migrate while app roles hold CREATE on the schema. cluster .owner @@ -613,19 +636,17 @@ async fn migration_repairs_lanes_it_may_and_reports_the_rest() -> TestResult { )))) .await?; - // The migration runs as the migration role itself. It is a member of the - // owner, so it may reassign its own and the owner's sequences, but not - // the app role's. - for (_, _, statement) in migrations::migration_sql_range(52, migrations::CURRENT_VERSION) { - cluster - .migrator - .execute(sqlx::raw_sql(sql(statement))) - .await?; - } + // `awa migrate` connects as the migration role and acts as the owner. As + // the owner it adds the grants the owner's sequences lack; the migration + // role's own sequences need the migration role's privileges, so they are + // reassigned as the login role. Neither can act as the app role. + let migrate_pool = cluster.migrator_as_owner().await?; + tokio::time::timeout(Duration::from_secs(120), migrations::run(&migrate_pool)).await??; assert_eq!( migrations::current_version(&cluster.owner).await?, migrations::CURRENT_VERSION ); + migrate_pool.close().await; cluster .assert_lanes_owned_by_owner("migrator_lane", 1) .await?; @@ -633,10 +654,10 @@ async fn migration_repairs_lanes_it_may_and_reports_the_rest() -> TestResult { assert_eq!(drain(&cluster.app_b, "migrator_lane").await?, 1); assert_eq!(drain(&cluster.app_b, "owner_lane").await?, 1); - let problems = storage::lane_sequence_problems(&cluster.app_c, "awa").await?; - assert_eq!(problems.len(), 2, "{problems:?}"); let app_a = cluster.roles[3].clone(); let app_a = &app_a; + let problems = storage::lane_sequence_problems(&cluster.app_c, "awa").await?; + assert_eq!(problems.len(), 2, "{problems:?}"); for problem in &problems { assert_eq!(&problem.owner, app_a); assert_eq!(problem.expected_owner, cluster.owner_name); @@ -657,21 +678,31 @@ async fn migration_repairs_lanes_it_may_and_reports_the_rest() -> TestResult { .as_str() ) ); + assert!(problem.repair_roles.is_empty(), "{problem:?}"); } + let next_steps = + storage::lane_sequence_next_steps(problems.iter().map(|problem| (problem, false))); + assert_eq!(next_steps.len(), 1, "{next_steps:?}"); + assert!( + next_steps[0].contains(&format!("no login role can both act as {app_a}")), + "{next_steps:?}" + ); // Re-running the repair as the migration role changes nothing and fails - // nothing. The migration role created the repair function, so it grants - // its use. + // nothing. cluster - .migrator + .owner .execute(sqlx::raw_sql(sql(format!( - "GRANT EXECUTE ON FUNCTION awa.repair_lane_sequences(text) TO {}", + "GRANT EXECUTE ON FUNCTION awa.repair_lane_sequence(text, text) TO {}", cluster.runtime_name )))) .await?; - let rerun = storage::repair_lane_sequences(&cluster.migrator, "awa").await?; + let rerun = + storage::repair_lane_sequences(&mut *cluster.migrator.acquire().await?, "awa").await?; assert_eq!(rerun.len(), 2); - assert!(rerun.iter().all(|result| result.outcome == "unrepaired")); + assert!(rerun + .iter() + .all(|result| result.outcome == LaneSequenceOutcome::Unrepaired)); // A role that can act as the app role but cannot become the owner adds // the grants and leaves the owner for someone who can. @@ -686,20 +717,14 @@ async fn migration_repairs_lanes_it_may_and_reports_the_rest() -> TestResult { .await?; let delegate_pool = PgPoolOptions::new() .max_connections(1) - .connect_with( - PgConnectOptions::from_str(&std::env::var("DATABASE_URL").unwrap_or_else(|_| { - "postgres://postgres:test@localhost:15432/awa_test".to_string() - }))? - .database(&cluster.database) - .username(&delegate) - .password("lane_test"), - ) + .connect_with(cluster.connect_options.clone().username(&delegate)) .await?; - let delegated = storage::repair_lane_sequences(&delegate_pool, "awa").await?; + let delegated = + storage::repair_lane_sequences(&mut *delegate_pool.acquire().await?, "awa").await?; delegate_pool.close().await; assert_eq!(delegated.len(), 2); for result in &delegated { - assert_eq!(result.outcome, "owner_unchanged"); + assert_eq!(result.outcome, LaneSequenceOutcome::OwnerUnchanged); assert_eq!(&result.owner, app_a); assert!(result.missing_grants.is_empty()); assert!(result.reassign_sql.is_some()); @@ -707,7 +732,8 @@ async fn migration_repairs_lanes_it_may_and_reports_the_rest() -> TestResult { assert_eq!(drain(&cluster.app_b, "app_lane").await?, 1); enqueue(&cluster.app_c, "app_lane", 1, 1).await?; - // A role that is a member of both the app role and the owner reassigns. + // Once the migration role is a member of the app role too, the check names + // it, and a repair connected as it reassigns. cluster .admin .execute(sqlx::raw_sql(sql(format!( @@ -715,21 +741,93 @@ async fn migration_repairs_lanes_it_may_and_reports_the_rest() -> TestResult { migrator = cluster.migrator_name )))) .await?; - let repaired = storage::repair_lane_sequences(&cluster.migrator, "awa").await?; + let problems = storage::lane_sequence_problems(&cluster.app_c, "awa").await?; + assert!(problems + .iter() + .all(|problem| problem.repair_roles == vec![cluster.migrator_name.clone()])); + let repaired = + storage::repair_lane_sequences(&mut *cluster.migrator.acquire().await?, "awa").await?; assert_eq!(repaired.len(), 2); assert!( - repaired.iter().all(|result| result.outcome == "repaired"), + repaired + .iter() + .all(|result| result.outcome == LaneSequenceOutcome::Repaired), "{repaired:?}" ); assert!(storage::lane_sequence_problems(&cluster.app_c, "awa") .await? .is_empty()); - assert!(storage::repair_lane_sequences(&cluster.migrator, "awa") - .await? - .is_empty()); cluster.assert_lanes_owned_by_owner("app_lane", 1).await?; assert_eq!(drain(&cluster.app_b, "app_lane").await?, 1); cluster.roles.push(delegate); cluster.teardown().await } + +/// Reassigning a lane sequence waits on any open enqueue or claim on the lane, +/// so the repair bounds that wait and holds each sequence's lock only for its +/// own short transaction. A held sequence is skipped and reported; the rest +/// are repaired, and a later repair finishes it. +#[tokio::test] +async fn migration_skips_a_lane_sequence_an_open_transaction_holds() -> TestResult { + let cluster = Cluster::new(Schema::BeforeProvisioning, true).await?; + enqueue(&cluster.migrator, "held_lane", 1, 1).await?; + enqueue(&cluster.migrator, "free_lane", 1, 1).await?; + let held_sequence = cluster.enqueue_sequence("held_lane").await?; + // 0.7 refuses to migrate while app roles hold CREATE on the schema. + cluster + .owner + .execute(sqlx::raw_sql(sql(format!( + "REVOKE CREATE ON SCHEMA awa FROM {}", + cluster.runtime_name + )))) + .await?; + + // An enqueue in flight on held_lane holds its sequence's lock. + let mut holder = cluster.app_b.begin().await?; + sqlx::query("SELECT nextval($1::regclass)") + .bind(format!("awa.{held_sequence}")) + .execute(&mut *holder) + .await?; + + let migrate_pool = cluster.migrator_as_owner().await?; + // The migration completes while the lock is held. + tokio::time::timeout(Duration::from_secs(120), migrations::run(&migrate_pool)).await??; + for (sequence, owner) in cluster.lane_sequence_owners("held_lane").await? { + let expected = if sequence == held_sequence { + &cluster.migrator_name + } else { + &cluster.owner_name + }; + assert_eq!(&owner, expected, "{sequence}"); + } + cluster.assert_lanes_owned_by_owner("free_lane", 1).await?; + + // A repair while the transaction is still open reports the sequence busy. + let mut conn = migrate_pool.acquire().await?; + let busy = storage::repair_lane_sequences(&mut conn, "awa").await?; + assert_eq!(busy.len(), 1, "{busy:?}"); + assert_eq!(busy[0].sequence_name, held_sequence); + assert_eq!(busy[0].outcome, LaneSequenceOutcome::Busy); + assert_eq!(busy[0].owner, cluster.migrator_name); + let next_steps = storage::lane_sequence_repair_next_steps(&busy); + assert_eq!(next_steps.len(), 1); + assert!(next_steps[0].contains("busy"), "{next_steps:?}"); + assert!( + next_steps[0].contains("awa storage lane-sequences --schema awa --repair"), + "{next_steps:?}" + ); + + holder.rollback().await?; + let repaired = storage::repair_lane_sequences(&mut conn, "awa").await?; + assert_eq!(repaired.len(), 1, "{repaired:?}"); + assert_eq!(repaired[0].outcome, LaneSequenceOutcome::Repaired); + drop(conn); + migrate_pool.close().await; + cluster.assert_lanes_owned_by_owner("held_lane", 1).await?; + assert!(storage::lane_sequence_problems(&cluster.app_c, "awa") + .await? + .is_empty()); + assert_eq!(drain(&cluster.app_b, "held_lane").await?, 1); + cluster.teardown().await +} diff --git a/docs/security/database-roles.md b/docs/security/database-roles.md index 5f15793e7..d8d9f5497 100644 --- a/docs/security/database-roles.md +++ b/docs/security/database-roles.md @@ -85,17 +85,32 @@ awa --database-url "$AWA_DATABASE_URL" storage lane-sequences --schema awa SELECT * FROM awa.lane_sequence_problems('awa'); ``` -The command exits 2 while any remain. Migration v053 runs the repair as the migrating role: for each sequence it changes the owner to the schema owner and adds the missing grants, wherever that role is allowed to. A sequence it is not allowed to change does not fail the migration. It is reported in a warning, and `awa migrate` prints a summary. Run the repair again at any time; it is idempotent: +The command exits 2 while any remain. + +`awa migrate` repairs them once the migration commits, one sequence per short transaction: it gives each sequence to the schema owner and adds its missing grants, wherever it is allowed to. Reassigning takes the sequence's `ACCESS EXCLUSIVE` lock, so the repair waits at most 300 ms for it. A sequence that an open enqueue or claim holds for longer is skipped, left unchanged and reported as busy. A sequence the repair is not allowed to change is reported with the roles that can finish it. Neither fails the migration. Run the repair again at any time; it is idempotent: ```bash -PGOPTIONS='-c role=awa_owner' \ - awa --database-url "$AWA_MIGRATOR_DATABASE_URL" storage lane-sequences --repair +awa --database-url "$AWA_MIGRATOR_DATABASE_URL" storage lane-sequences --repair ``` -Changing a sequence's owner requires a role that can `SET ROLE` to both the current owner and the schema owner. A migrator usually can for sequences it or the owner created, but not for one an unrelated application role created, and managed PostgreSQL offers no superuser. For each sequence that remains, the output gives the statements to run: +Which role can repair a sequence: + +- Reassigning it needs a role that holds the current owner's privileges (a member of the owner, with inheritance) and can `SET ROLE` to the schema owner. +- Adding its grants needs the current owner's privileges. + +The schema owner holds neither the migration role's nor an application role's privileges, so a repair running as `awa_owner` cannot reassign sequences those roles created. That includes the migration role connected with `PGOPTIONS='-c role=awa_owner'`. Run the repair as the migration role itself, without `PGOPTIONS`, as above. `awa migrate` and `--repair` retry each sequence the effective role could not repair as the login role, so `awa migrate` run with `PGOPTIONS='-c role=awa_owner'`, as documented, still repairs the migration role's sequences. + +For each sequence that remains, the output lists `repair_roles`, the login roles other than superusers that can repair it, and the statements to run. A busy sequence only needs another `--repair`. When `repair_roles` is empty, which is usual for a sequence an application role created on managed PostgreSQL, either: -- `missing_grants`: run them as the current owner, or as a role that can `SET ROLE` to it. Every runtime role can then use the lane, although the owner stays unchanged. -- `reassign_sql`: run it as a role that is a member of both roles, for example after `GRANT app_role TO awa_migrator` from a role with `ADMIN` on `app_role`. Revoke that membership afterwards. +- run `missing_grants` as the current owner. Every runtime role can then use the lane, although the owner stays unchanged; or +- give the migration role the owner's privileges, for example `GRANT app_role TO awa_migrator` from a role with `ADMIN` on `app_role`, rerun `--repair` as the migration role, then revoke the membership. + +Without the CLI, repair one sequence per transaction from `psql` in autocommit mode, connected as the migration role: + +```sql +SELECT format('SELECT * FROM awa.repair_lane_sequence(%L, %L)', 'awa', sequence_name) +FROM awa.lane_sequence_problems('awa') \gexec +``` ## Custom queue-storage schemas ## Custom queue-storage schemas diff --git a/docs/upgrade-0.6-to-0.7.md b/docs/upgrade-0.6-to-0.7.md index 7fff6173f..59ef9ba89 100644 --- a/docs/upgrade-0.6-to-0.7.md +++ b/docs/upgrade-0.6-to-0.7.md @@ -28,9 +28,10 @@ default privileges covered that role could use the lane; any other role failed with `permission denied for sequence`. v053 installs `SECURITY DEFINER` `provision_lane` and `provision_lane_sequences` functions, owned by the schema owner, in each queue-storage schema. The lane helpers create sequences through -them, with grants that mirror the lane's head table. v053 then gives every -existing lane sequence to the schema owner and adds its missing grants, as far -as the migrating role is allowed to. +them, with grants that mirror the lane's head table. Once v053 commits, +`awa migrate` gives every existing lane sequence to the schema owner and adds +its missing grants, as far as it is allowed to, one sequence per short +transaction with a bounded lock wait. 0.6 and earlier 0.7 binaries call the same helpers, so lanes they create after v053 are owned by the schema owner too. A 0.7 binary running ahead of v053 @@ -42,13 +43,15 @@ To adopt it: while lanes need repair). 2. Run `awa migrate` with the schema owner as the effective role, as described in [Database roles](security/database-roles.md). -3. Read the migrate output. Any lane sequence it could not reassign is listed; - `awa storage lane-sequences` gives the statements to run and the role to run - them as. See +3. Read the migrate output. Any lane sequence it could not repair is listed in + a warning. A busy one needs another run of + `awa storage lane-sequences --repair` as the migration role (without + `PGOPTIONS`); the others name the roles that can repair them. See [Check and repair lane ownership](security/database-roles.md#check-and-repair-lane-ownership). External migration runners apply the generated `awa migrate --sql --from 52 ---to 53` bundle, then run `awa storage lane-sequences` to check for leftovers. +--to 53` bundle, which does not repair existing lane sequences, then run +`awa storage lane-sequences --repair` as the migration role itself. ## Custom-schema discovery during migration diff --git a/skills/awa-operations/SKILL.md b/skills/awa-operations/SKILL.md index 5124fbc3f..64a23de16 100644 --- a/skills/awa-operations/SKILL.md +++ b/skills/awa-operations/SKILL.md @@ -96,7 +96,8 @@ From v053 any runtime role creates lanes through the schema owner's CREATE nor sequence default privileges, and `awa storage prepare-queue` is optional. Run migrations with the schema owner as the effective role. Check lane ownership with `awa storage lane-sequences` (exits 2 while lanes need -repair) and fix it with `--repair` or the statements it prints; see +repair) and fix it with `--repair`, connected as the migration role itself +rather than as the schema owner, or with the statements it prints; see [Database roles](../../docs/security/database-roles.md#check-and-repair-lane-ownership). ## Rolling Upgrades From bfccec55abf279ce6ac9c1344b3f77cd1ed29490 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Fri, 6 Nov 2026 20:58:37 +1300 Subject: [PATCH 5/5] docs(security): clarify lane sequence repair outcomes --- awa-model/migrations/v053_definer_lane_provisioning.sql | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/awa-model/migrations/v053_definer_lane_provisioning.sql b/awa-model/migrations/v053_definer_lane_provisioning.sql index 2eba66ff6..ced1a7642 100644 --- a/awa-model/migrations/v053_definer_lane_provisioning.sql +++ b/awa-model/migrations/v053_definer_lane_provisioning.sql @@ -45,7 +45,7 @@ BEGIN IF v_count > 0 THEN RAISE NOTICE '% lane sequence(s) in schema % are not owned by the schema owner or lack runtime grants', v_count, v_schema USING HINT = format( - 'awa migrate repairs them after this commits. Otherwise run awa storage lane-sequences --schema %s --repair as the migration role itself, without SET ROLE.', + 'awa migrate attempts repair after this commits; review its warnings for any sequences left unrepaired. To retry, run awa storage lane-sequences --schema %s --repair as the migration role itself, without SET ROLE.', v_schema ); END IF;